04-MySQL实战¶

import pymysql

# 建立连接
conn = pymysql.connect(
    host='localhost',
    port=3306,
    user='root',
    password='your_password',
    database='mydb',
    charset='utf8mb4'
)

try:
    # 创建游标
    with conn.cursor() as cursor:
        # 执行SQL
        cursor.execute("SELECT VERSION()")
        result = cursor.fetchone()
        print(f"MySQL版本: {result[0]}")

        # 创建表
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INT PRIMARY KEY AUTO_INCREMENT,
                username VARCHAR(50) NOT NULL,
                email VARCHAR(100) UNIQUE,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        # 插入数据
        cursor.execute(
            "INSERT INTO users (username, email) VALUES (%s, %s)",
            ('张三', 'zhangsan@example.com')
        )

        # 批量插入
        users = [
            ('李四', 'lisi@example.com'),
            ('王五', 'wangwu@example.com'),
            ('赵六', 'zhaoliu@example.com')
        ]
        cursor.executemany(
            "INSERT INTO users (username, email) VALUES (%s, %s)",
            users
        )

        # 提交事务
        conn.commit()

        # 查询数据
        # ⚠️ 生产环境应避免 SELECT *，明确指定需要的列（如 SELECT id, username, email）
        cursor.execute("SELECT * FROM users WHERE id > %s", (0,))
        results = cursor.fetchall()

        for row in results:
            print(f"ID: {row[0]}, 用户名: {row[1]}, 邮箱: {row[2]}")

        # 查询单条
        cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
        user = cursor.fetchone()
        print(f"用户: {user}")

        # 更新数据
        cursor.execute(
            "UPDATE users SET email = %s WHERE id = %s",
            ('newemail@example.com', 1)
        )
        conn.commit()

        # 删除数据
        cursor.execute("DELETE FROM users WHERE id = %s", (5,))
        conn.commit()

finally:
    conn.close()

import pymysql
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    """数据库连接上下文管理器"""
    conn = pymysql.connect(
        host='localhost',
        user='root',
        password='your_password',
        database='mydb',
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor  # 返回字典格式
    )
    try:
        yield conn
    finally:
        conn.close()

@contextmanager
def get_cursor(conn):
    """游标上下文管理器"""
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cursor.close()

# 使用示例
with get_db_connection() as conn:
    with get_cursor(conn) as cursor:
        cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
        user = cursor.fetchone()
        print(user)  # {'id': 1, 'username': '张三', ...}

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker, declarative_base
from datetime import datetime

# 创建引擎
engine = create_engine(
    'mysql+pymysql://root:password@localhost:3306/mydb?charset=utf8mb4',
    echo=True  # 打印SQL语句
)

# 声明基类
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)
    created_at = Column(DateTime, default=datetime.now)

    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"

# 创建表
Base.metadata.create_all(engine)

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 增
try:
    new_user = User(username='张三', email='zhangsan@example.com')
    session.add(new_user)
    session.commit()
    print(f"创建用户: {new_user.id}")
except Exception as e:
    session.rollback()
    print(f"错误: {e}")

# 查
user = session.query(User).filter_by(username='张三').first()
print(f"查询结果: {user}")

# 查所有
users = session.query(User).all()
for u in users:
    print(u)

# 条件查询
users = session.query(User).filter(User.id > 0).order_by(User.created_at.desc()).limit(10).all()

# 改
user = session.query(User).filter_by(username='张三').first()
if user:
    user.email = 'newemail@example.com'
    session.commit()

# 删
user = session.query(User).filter_by(username='张三').first()
if user:
    session.delete(user)
    session.commit()

# 关闭会话
session.close()

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

# 使用连接池
engine = create_engine(
    'mysql+pymysql://root:password@localhost:3306/mydb',
    poolclass=QueuePool,
    pool_size=10,           # 连接池大小
    max_overflow=20,        # 最大溢出连接
    pool_recycle=3600,      # 连接回收时间（秒）
    pool_pre_ping=True      # 连接前ping测试
)

Session = sessionmaker(bind=engine)

import pymysql
import hashlib
import secrets
import json
from datetime import datetime
from contextlib import contextmanager

class UserManager:
    def __init__(self, host='localhost', user='root', password='', database='mydb'):
        self.config = {
            'host': host,
            'user': user,
            'password': password,
            'database': database,
            'charset': 'utf8mb4',
            'cursorclass': pymysql.cursors.DictCursor
        }

    @contextmanager
    def _get_connection(self):
        conn = pymysql.connect(**self.config)
        try:  # try/except捕获异常
            yield conn  # yield生成器：惰性产出值，节省内存
        finally:
            conn.close()

    def _hash_password(self, password: str, salt: str = None) -> tuple:
        """密码哈希"""
        if salt is None:
            salt = secrets.token_hex(16)
        hash_value = hashlib.pbkdf2_hmac('sha256', password.encode(), salt.encode(), 100000)
        return hash_value.hex(), salt

    def _verify_password(self, password: str, hash_value: str, salt: str) -> bool:
        """验证密码"""
        computed_hash, _ = self._hash_password(password, salt)
        return computed_hash == hash_value

    def register(self, username: str, password: str, email: str = None, phone: str = None) -> dict:
        """用户注册"""
        password_hash, salt = self._hash_password(password)
        full_hash = f"{salt}${password_hash}"

        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                try:
                    cursor.execute(
                        "INSERT INTO users (username, password_hash, email, phone) VALUES (%s, %s, %s, %s)",
                        (username, full_hash, email, phone)
                    )
                    user_id = cursor.lastrowid

                    # 分配默认角色
                    cursor.execute(
                        "INSERT INTO user_roles (user_id, role_id) VALUES (%s, (SELECT role_id FROM roles WHERE role_name = 'user'))",
                        (user_id,)
                    )

                    conn.commit()

                    self._log_operation(user_id, 'REGISTER', 'user', str(user_id), {'username': username})

                    return {'success': True, 'user_id': user_id, 'message': '注册成功'}

                except pymysql.err.IntegrityError as e:
                    if 'username' in str(e):
                        return {'success': False, 'message': '用户名已存在'}
                    elif 'email' in str(e):
                        return {'success': False, 'message': '邮箱已被使用'}
                    raise

    def login(self, username: str, password: str, ip_address: str = None) -> dict:
        """用户登录"""
        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute(
                    "SELECT user_id, password_hash, status FROM users WHERE username = %s",
                    (username,)
                )
                user = cursor.fetchone()

                if not user:
                    return {'success': False, 'message': '用户名或密码错误'}

                if user['status'] == 'banned':
                    return {'success': False, 'message': '账号已被禁用'}

                # 验证密码
                salt, hash_value = user['password_hash'].split('$')
                if not self._verify_password(password, hash_value, salt):
                    self._log_operation(user['user_id'], 'LOGIN_FAILED', 'user', str(user['user_id']),
                                      {'username': username, 'ip': ip_address})
                    return {'success': False, 'message': '用户名或密码错误'}

                # 获取用户角色
                cursor.execute("""
                    SELECT r.role_name FROM roles r
                    JOIN user_roles ur ON r.role_id = ur.role_id
                    WHERE ur.user_id = %s
                """, (user['user_id'],))
                roles = [row['role_name'] for row in cursor.fetchall()]

                self._log_operation(user['user_id'], 'LOGIN', 'user', str(user['user_id']),
                                  {'username': username, 'ip': ip_address})

                return {
                    'success': True,
                    'user_id': user['user_id'],
                    'username': username,
                    'roles': roles,
                    'message': '登录成功'
                }

    def get_user(self, user_id: int) -> dict:
        """获取用户信息"""
        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute(
                    "SELECT user_id, username, email, phone, status, created_at FROM users WHERE user_id = %s",
                    (user_id,)
                )
                return cursor.fetchone()

    def update_user(self, user_id: int, **kwargs) -> dict:  # *args接收任意位置参数；**kwargs接收任意关键字参数
        """更新用户信息"""
        allowed_fields = ['email', 'phone', 'status']
        updates = {k: v for k, v in kwargs.items() if k in allowed_fields}

        if not updates:
            return {'success': False, 'message': '没有可更新的字段'}

        set_clause = ', '.join([f"{k} = %s" for k in updates.keys()])
        values = list(updates.values()) + [user_id]

        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute(f"UPDATE users SET {set_clause} WHERE user_id = %s", values)
                conn.commit()

                self._log_operation(user_id, 'UPDATE_USER', 'user', str(user_id), updates)

                return {'success': True, 'message': '更新成功'}

    def delete_user(self, user_id: int) -> dict:
        """删除用户"""
        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute("DELETE FROM users WHERE user_id = %s", (user_id,))
                conn.commit()

                self._log_operation(user_id, 'DELETE_USER', 'user', str(user_id), {})

                return {'success': True, 'message': '删除成功'}

    def list_users(self, page: int = 1, page_size: int = 10, status: str = None) -> dict:
        """用户列表"""
        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                where_clause = "WHERE status = %s" if status else ""
                params = [status] if status else []

                # 查询总数
                cursor.execute(f"SELECT COUNT(*) as total FROM users {where_clause}", params)
                total = cursor.fetchone()['total']

                # 查询数据
                offset = (page - 1) * page_size
                cursor.execute(f"""
                    SELECT user_id, username, email, phone, status, created_at
                    FROM users {where_clause}
                    ORDER BY created_at DESC
                    LIMIT %s OFFSET %s
                """, params + [page_size, offset])

                users = cursor.fetchall()

                return {
                    'success': True,
                    'total': total,
                    'page': page,
                    'page_size': page_size,
                    'users': users
                }

    def _log_operation(self, user_id: int, operation: str, target_type: str,
                      target_id: str, details: dict, ip_address: str = None):
        """记录操作日志"""
        with self._get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute(
                    "INSERT INTO operation_logs (user_id, operation, target_type, target_id, details, ip_address) VALUES (%s, %s, %s, %s, %s, %s)",
                    (user_id, operation, target_type, target_id, json.dumps(details), ip_address)  # json.dumps将Python对象转为JSON字符串
                )
                conn.commit()

# 使用示例
if __name__ == '__main__':
    um = UserManager(password='your_password')

    # 注册
    result = um.register('testuser', 'password123', 'test@example.com')
    print(result)

    # 登录
    result = um.login('testuser', 'password123')
    print(result)

    # 获取用户列表
    result = um.list_users(page=1, page_size=5)
    print(result)