SQLAlchemy 使用详解

SQLAlchemy 是 Python 中最强大的 ORM（对象关系映射）工具之一，提供了强大的查询功能、事务管理和数据映射能力。本文将介绍 SQLAlchemy 的通用使用方法，帮助开发者构建健壮的数据访问层。

1. 架构概述

使用 SQLAlchemy 时，通常采用以下架构模式：

数据访问层架构
+---------------------+
|   实体层 (Entities)  |
|   定义业务对象结构   |
+---------------------+
          ↓
+---------------------+
|   映射层 (Mappers)   |
| 实体与数据库模型映射 |
+---------------------+
          ↓
+---------------------+
|  数据库模型 (Models)  |
|  SQLAlchemy 映射类   |
+---------------------+
          ↓
+---------------------+
|   存储库 (Repositories) |
| 数据访问操作接口   |
+---------------------+
          ↓
+---------------------+
| 连接管理器 (Client Manager) |
| 数据库连接与会话管理 |
+---------------------+

2. SQLAlchemy 核心组件

2.1 连接管理

from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine, AsyncSession, async_sessionmaker

class MySQLClientManager:
    def __init__(self, config: dict):
        self.engine: AsyncEngine | None = None
        self.session_factory = None
        self.config = config

    def _get_url(self):
        base_url = f"mysql+asyncmy://{self.config['user']}:{self.config['password']}@" \
                  f"{self.config['host']}:{self.config['port']}"
        if self.config.get('database'):
            base_url += f"/{self.config['database']}"
        base_url += "?charset=utf8mb4"
        return base_url

    def init(self):
        self.engine = create_async_engine(self._get_url(), pool_size=10, pool_pre_ping=True)
        self.session_factory = async_sessionmaker(self.engine, autoflush=True, expire_on_commit=False)

    async def close(self):
        await self.engine.dispose()

关键特性：

使用 sqlalchemy.ext.asyncio 实现异步操作
配置连接池管理（pool_size=10）
启用连接检查（pool_pre_ping=True）
支持灵活的数据库连接配置

2.2 基础模型类

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

说明：

定义了 SQLAlchemy 的基础模型类
所有数据库模型都继承自此类
使用 DeclarativeBase 简化模型定义

2.3 数据库模型

from sqlalchemy import String, Text, Integer, DateTime
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime

class TableInfo(Base):
    __tablename__ = "table_info"
    __table_args__ = {"extend_existing": True}

    id: Mapped[str] = mapped_column(
        String(64),
        primary_key=True,
        comment="表编号"
    )
    name: Mapped[str | None] = mapped_column(
        String(128),
        comment="表名称"
    )
    role: Mapped[str | None] = mapped_column(
        String(32),
        comment="表类型(fact/dim)"
    )
    description: Mapped[str | None] = mapped_column(
        Text,
        comment="表描述"
    )
    database: Mapped[str | None] = mapped_column(
        String(64),
        default="default_db",
        comment="所属数据库名称"
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        default=datetime.now,
        comment="创建时间"
    )

模型特性：

使用类型注解风格的 Mapped 和 mapped_column
支持非空字段和默认值
提供数据库字段注释
配置表扩展属性 (extend_existing)

3. 数据访问与操作

3.1 实体与模型映射

项目采用数据映射模式，将业务实体与数据库模型分离：

class TableInfoEntity:
    """表信息实体类"""
    
    def __init__(self, id: str, name: str = None, role: str = None, 
                 description: str = None, database: str = None):
        self.id = id
        self.name = name
        self.role = role
        self.description = description
        self.database = database


class TableInfoMapper:
    """TableInfo 实体与模型之间的映射类"""
    
    @staticmethod
    def to_model(entity: TableInfoEntity) -> TableInfo:
        """将实体转换为数据库模型"""
        return TableInfo(
            id=entity.id,
            name=entity.name,
            role=entity.role,
            description=entity.description,
            database=entity.database
        )
    
    @staticmethod
    def to_entity(model: TableInfo) -> TableInfoEntity:
        """将数据库模型转换为实体"""
        return TableInfoEntity(
            id=model.id,
            name=model.name,
            role=model.role,
            description=model.description,
            database=model.database
        )

3.2 存储库模式

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

class MySQLRepository:
    """
    MySQL 通用存储库类

    负责与 MySQL 数据库交互，提供通用的数据访问接口。
    """

    def __init__(self, session: AsyncSession):
        """
        初始化 MySQL 存储库

        Args:
            session: SQLAlchemy 异步会话对象
        """
        self.session = session

    async def save_entity(self, entity: TableInfoEntity):
        """
        保存或更新实体到 MySQL 数据库

        Args:
            entity: 实体对象
        """
        existing = await self.session.get(TableInfo, entity.id)
        if existing:
            # 如果已存在，更新信息
            existing.name = entity.name
            existing.role = entity.role
            existing.description = entity.description
            existing.database = entity.database
        else:
            # 如果不存在，插入新记录
            self.session.add(TableInfoMapper.to_model(entity))

    async def get_entity_by_id(self, id: str) -> TableInfoEntity | None:
        """
        根据 ID 查询实体

        Args:
            id: 实体 ID

        Returns:
            TableInfoEntity | None: 实体对象，若不存在则返回 None
        """
        model = await self.session.get(TableInfo, id)
        if model:
            return TableInfoMapper.to_entity(model)
        else:
            return None

    async def get_all_entities(self) -> list[TableInfoEntity]:
        """
        查询所有实体

        Returns:
            list[TableInfoEntity]: 实体列表
        """
        from sqlalchemy.future import select
        result = await self.session.execute(select(TableInfo))
        return [TableInfoMapper.to_entity(row[0]) for row in result]

关键特性：

提供统一的数据访问接口
支持实体与模型之间的自动映射
实现了常用的 CRUD 操作
使用 SQLAlchemy 会话管理事务

4. 会话管理

在实际使用中，我们使用异步会话工厂来管理数据库会话：

import asyncio
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

# 配置数据库连接
config = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "password",
    "database": "mydatabase"
}

# 初始化连接管理器
manager = MySQLClientManager(config)
manager.init()

async def test_query():
    async with manager.session_factory() as session:
        try:
            repo = MySQLRepository(session)
            
            # 测试查询所有表信息
            tables = await repo.get_all_entities()
            print(f"查询到 {len(tables)} 条表信息")
            if tables:
                print(f"第一条记录: {tables[0].name}")
        except Exception as e:
            print(f"查询失败: {e}")

asyncio.run(test_query())

会话管理模式：

使用异步上下文管理器自动管理会话
确保会话正确关闭和资源释放
支持事务管理

5. 高级查询功能

SQLAlchemy 提供了强大的查询 API，支持各种查询操作：

from sqlalchemy.future import select

# 查询所有表信息
async def get_all_tables():
    async with manager.session_factory() as session:
        repo = MySQLRepository(session)
        return await repo.get_all_entities()

# 查询特定类型的表
async def get_specific_tables(role: str):
    async with manager.session_factory() as session:
        result = await session.execute(
            select(TableInfo).where(TableInfo.role == role)
        )
        return [TableInfoMapper.to_entity(row[0]) for row in result]

# 使用原生 SQL 查询
async def get_tables_with_sql():
    async with manager.session_factory() as session:
        sql = "select * from table_info where role in ('fact', 'dim')"
        result = await session.execute(text(sql))
        return [TableInfo(**dict(row)) for row in result.mappings().fetchall()]

6. 性能优化技巧

6.1 连接池配置

def init(self):
    self.engine = create_async_engine(
        self._get_url(),
        pool_size=20,
        max_overflow=10,
        pool_recycle=3600,
        pool_pre_ping=True
    )
    self.session_factory = async_sessionmaker(self.engine, autoflush=True, expire_on_commit=False)

优化建议：

根据实际并发量调整 pool_size
使用 max_overflow 处理突发请求
设置 pool_recycle 防止连接过期
使用 pool_pre_ping 确保连接有效性

6.2 查询优化

# 批量查询示例
async def get_tables_by_ids(table_ids: list[str]) -> list[TableInfoEntity]:
    async with manager.session_factory() as session:
        result = await session.execute(
            select(TableInfo).where(TableInfo.id.in_(table_ids))
        )
        return [TableInfoMapper.to_entity(row[0]) for row in result]

7. 错误处理与调试

7.1 错误处理

import logging

async def safe_query(query_func):
    """安全查询装饰器"""
    async def wrapper(*args, **kwargs):
        try:
            return await query_func(*args, **kwargs)
        except Exception as e:
            logging.error(f"查询失败: {str(e)}", exc_info=True)
            return None
    return wrapper

@safe_query
async def get_safe_table_info(table_id: str) -> TableInfoEntity:
    """安全查询表信息"""
    async with manager.session_factory() as session:
        repo = MySQLRepository(session)
        return await repo.get_entity_by_id(table_id)

7.2 查询调试

使用 SQLAlchemy 的查询日志功能：

import logging

# 配置 SQLAlchemy 日志
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# 查询时会打印 SQL 语句
async with manager.session_factory() as session:
    result = await session.execute(select(TableInfo))
    print(f"查询到 {result.scalar_one_or_none()} 条记录")

8. 最佳实践

8.1 分层架构

遵循清晰的分层架构：

实体层：定义业务对象结构
映射层：处理实体与模型的转换
存储库层：提供数据访问接口
服务层：处理业务逻辑

8.2 会话管理

始终使用上下文管理器管理会话
避免在长时间运行的操作中保持会话
合理设置会话超时时间

8.3 查询优化

避免 N+1 查询问题，使用 selectinload 或 joinedload 预加载
合理使用索引
避免不必要的字段查询，只查询需要的数据

8.4 事务管理

使用异步上下文管理器处理事务
确保事务边界正确
处理事务异常和回滚

9. 总结

SQLAlchemy 提供了一个强大、灵活且可靠的数据库访问层。通过使用异步操作、连接池管理和高级查询功能，我们能够处理各种复杂的数据访问需求。同时，采用实体-模型映射和存储库模式，使代码结构清晰、易于维护和扩展。

在实际项目中，还可以根据具体需求进一步优化查询性能、增加缓存策略和监控机制，以确保数据访问层的高可用性和响应性。

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

博客园 - work hard work smart