#A complete guide to connecting Python to MySQL and PostgreSQL databases

# config.py
from dotenv import load_dotenv
import os

load_dotenv()  # 自动读取 .env 文件并注入环境变量

DB_CONFIG = {
    "mysql": {
        "host": os.getenv("MYSQL_HOST"),
        "port": int(os.getenv("MYSQL_PORT", 3306)),
        "user": os.getenv("MYSQL_USER"),
        "password": os.getenv("MYSQL_PASSWORD"),
        "database": os.getenv("MYSQL_DATABASE"),
        "charset": "utf8mb4",
        "autocommit": False,          # 显式控制事务
    },
    "postgresql": {
        "host": os.getenv("PG_HOST"),
        "port": int(os.getenv("PG_PORT", 5432)),
        "user": os.getenv("PG_USER"),
        "password": os.getenv("PG_PASSWORD"),
        "database": os.getenv("PG_DATABASE"),
    }
}

import logging
from contextlib import contextmanager

logging.basicConfig(level=logging.INFO)

# MySQL 上下文
def mysql_conn_context(config):
    import pymysql
    conn = None
    try:
        conn = pymysql.connect(**config)
        yield conn
        conn.commit()
    except Exception as e:
        if conn:
            conn.rollback()
        logging.error(f"MySQL 操作失败: {e}")
        raise
    finally:
        if conn:
            conn.close()

# PostgreSQL 上下文
def pg_conn_context(config):
    import psycopg2
    conn = None
    try:
        conn = psycopg2.connect(**config)
        yield conn
        conn.commit()
    except Exception as e:
        if conn:
            conn.rollback()
        logging.error(f"PostgreSQL 操作失败: {e}")
        raise
    finally:
        if conn:
            conn.close()

from config import DB_CONFIG

def create_mysql_user(name, email, age):
    """创建用户，并返回自增主键"""
    with mysql_conn_context(DB_CONFIG["mysql"]) as conn:
        cursor = conn.cursor()
        # 注意：参数化查询必须使用 %s 占位符
        sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
        cursor.execute(sql, (name, email, age))
        return cursor.lastrowid

def get_mysql_users(limit=10):
    """查询用户列表，结果以字典形式返回"""
    with mysql_conn_context(DB_CONFIG["mysql"]) as conn:
        # 指定 DictCursor 让结果更容易使用
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute("SELECT * FROM users LIMIT %s", (limit,))
        return cursor.fetchall()

from config import DB_CONFIG

def upsert_pg_user(data):
    """UPSERT 操作：插入，冲突则更新（PostgreSQL 独有）"""
    with pg_conn_context(DB_CONFIG["postgresql"]) as conn:
        cursor = conn.cursor()
        sql = """
            INSERT INTO users (name, email, age)
            VALUES (%(name)s, %(email)s, %(age)s)
            ON CONFLICT (email)                -- 要求 email 列有唯一索引
            DO UPDATE SET age = EXCLUDED.age
            RETURNING id, name, email
        """
        cursor.execute(sql, data)
        return cursor.fetchone()   # 直接拿到插入/更新后的行

from DBUtils.PooledDB import PooledDB
import pymysql

# 全局初始化连接池（程序启动时执行一次）
mysql_pool = PooledDB(
    creator=pymysql,           # 使用 PyMySQL 创建连接
    maxconnections=20,         # 最大允许连接数
    mincached=2,               # 池中初始空闲连接
    maxcached=5,               # 池中最大空闲连接
    **DB_CONFIG["mysql"]
)

# 封装为上下文管理器
def mysql_pool_context():
    conn = None
    try:
        conn = mysql_pool.connection()  # 从池中获取连接
        yield conn
        conn.commit()
    except Exception as e:
        if conn:
            conn.rollback()
        logging.error(f"MySQL 连接池操作失败: {e}")
        raise
    finally:
        if conn:
            conn.close()   # close 是归还连接，而非物理断开

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 构建 PostgreSQL 引擎，连接池默认启用
pg_engine = create_engine(
    "postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}".format(
        **DB_CONFIG["postgresql"]
    ),
    pool_size=10,           # 连接池保持的连接数
    max_overflow=10,        # 临时允许超出的连接数
    pool_pre_ping=True      # 每次取出连接前先探测是否有效
)

PgSession = sessionmaker(bind=pg_engine)

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), unique=True, nullable=False, index=True)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

# 创建表（生产环境请使用 Alembic 进行迁移管理）
Base.metadata.create_all(bind=pg_engine)

from contextlib import contextmanager

@contextmanager
def pg_orm_session():
    session = PgSession()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        logging.error(f"ORM 操作失败: {e}")
        raise
    finally:
        session.close()

# 创建用户
def create_orm_user(name, email, age):
    with pg_orm_session() as session:
        user = User(name=name, email=email, age=age)
        session.add(user)
        session.flush()   # 在不提交事务的情况下获取 ID
        return user.id

# 条件查询与排序
def get_young_users(min_age=18, max_age=30, limit=10):
    with pg_orm_session() as session:
        return session.query(User)\
                     .filter(User.age.between(min_age, max_age))\
                     .order_by(User.created_at.desc())\
                     .limit(limit)\
                     .all()

cursor.execute("SELECT * FROM users WHERE name = %s", (user_input_name,))

cursor.execute(f"SELECT * FROM users WHERE name = '{user_input_name}'")