Python ORM及SQLAlchemy学习总结（下）

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.pool import NullPool, QueuePool
from contextlib import asynccontextmanager
Base = declarative_base()
class AsyncDatabase:
    def init(self, db_url: str, pool_size: int = 10, max_overflow: int = 5):
        # 使用 QueuePool 连接池
        self.engine = create_async_engine(
            db_url,
            poolclass=QueuePool,
            pool_size=pool_size,
            max_overflow=max_overflow,
            pool_recycle=3600,  # 1小时回收连接
            echo=False,  # 生产环境设为 False
            future=True
        )
        self.async_session = sessionmaker(
            self.engine,
            class_=AsyncSession,
            expire_on_commit=False
        )

    @asynccontextmanager
    async def get_session(self):
        """异步上下文管理器获取数据库会话"""
        session = self.async_session()
        try:
            yield session
            await session.commit()
        except Exception as e:
            await session.rollback()
            raise
        finally:
            await session.close()

    async def create_all(self):
        """创建所有表结构"""
        async with self.engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
# PostgreSQL 示例
DB_URL = "postgresql+asyncpg://user:password@localhost/dbname"
# MySQL 示例
# DB_URL = "mysql+aiomysql://user:password@localhost/dbname"
# 全局数据库实例
db = AsyncDatabase(
    db_url=DB_URL,
    pool_size=15,
    max_overflow=5
)

# models.py
from sqlalchemy import Column, Integer, String
from .database import Base
class User(Base):
    tablename = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)

# services.py
from .database import db
from .models import User
from sqlalchemy import select, update, delete
from sqlalchemy.exc import SQLAlchemyError
from tornado.web import HTTPError
class UserService:
    @staticmethod
    async def create_user(name: str, email: str) -> User:
        async with db.get_session() as session:
            try:
                user = User(name=name, email=email)
                session.add(user)
                await session.flush()
                return user
            except SQLAlchemyError as e:
                raise HTTPError(500, f"Database error: {str(e)}")
    @staticmethod
    async def get_user(user_id: int) -> User:
        async with db.get_session() as session:
            stmt = select(User).where(User.id == user_id)
            result = await session.execute(stmt)
            user = result.scalar_one_or_none()
            if not user:
                raise HTTPError(404, "User not found")
            return user
    @staticmethod
    async def update_user(user_id: int, **kwargs) -> User:
        async with db.get_session() as session:
            try:
                stmt = (
                    update(User)
                    .where(User.id == user_id)
                    .values(**kwargs)
                    .returning(User)
                )
                result = await session.execute(stmt)
                user = result.scalar_one()
                return user
            except SQLAlchemyError as e:
                raise HTTPError(500, f"Update failed: {str(e)}")
    @staticmethod
    async def list_users(page: int = 1, per_page: int = 10) -> list[User]:
        async with db.get_session() as session:
            offset = (page - 1) * per_page
            stmt = select(User).offset(offset).limit(per_page)
            result = await session.execute(stmt)
            return result.scalars().all()

from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy import create_engine

engine = create_engine("sqlite:///:memory:")
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

# 在同一个线程中，多次调用Session()返回同一个Session对象
session1 = Session()
session2 = Session()
print(session1 is session2)  # 输出 True

# 使用完后，需要移除当前线程的Session
Session.remove()

from sqlalchemy.ext.asyncio import async_scoped_session, async_sessionmaker

async_session_factory = async_sessionmaker(engine, expire_on_commit=False)
AsyncScopedSession = async_scoped_session(async_session_factory, scopefunc=asyncio.current_task)

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
engine = create_async_engine("postgresql+asyncpg://...")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# 依赖项
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# 在路由中使用
@app.get("/")
async def read_root(session: AsyncSession = Depends(get_db)):
    ...