Python开发中，SQLAlchemy 的同步操作和异步操作封装，以及常规CRUD的处理。

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'  # 数据库表名
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

id = Column(Integer, primary_key=True)
name = Column(String(50))

new_user = User(id=1, name='John Doe', email='[email protected]')

id = Column(Integer, primary_key=True)

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship('User')

addresses = relationship("Address", back_populates="user")

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

session.add(new_user)
session.commit()

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from typing import Generator
from core.config import settings

# 常规同步处理
engine = create_engine(settings.DB_URI)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Generator:
    """创建一个 SQLAlchemy 数据库会话-同步处理."""
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

# mysql 数据库引擎
engine = create_engine(
    "mysql+pymysql://root:[email protected]:3306/WinFramework",
    pool_recycle=3600,
    # echo=True,
)

# Sqlite 数据库引擎
engine = create_engine("sqlite:///testdir//test.db")

# PostgreSQL 数据库引擎
engine = create_engine(
    "postgresql+psycopg2://postgres:123456@localhost:5432/winframework",
     # echo=True,
 )

# SQLServer 数据库引擎
engine = create_engine(
     "mssql+pymssql://sa:123456@localhost/WinFramework?tds_version=7.0",
     # echo=True,
 )

class CRUDOperations:
    def __init__(self, model):
        self.model = model

    def create(self, db, obj_in):
        db_obj = self.model(**obj_in.dict())
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def get(self, db, id):
        return db.query(self.model).filter(self.model.id == id).first()

    def update(self, db, db_obj, obj_in):
        obj_data = obj_in.dict(exclude_unset=True)
        for field in obj_data:
            setattr(db_obj, field, obj_data[field])
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def remove(self, db, id):
        obj = db.query(self.model).get(id)
        db.delete(obj)
        db.commit()
        return obj

crud_user = CRUDOperations(User)

# Create
with get_db() as db:
    user = crud_user.create(db, user_data)

# Read
with get_db() as db:
    user = crud_user.get(db, user_id)

# Update
with get_db() as db:
    user = crud_user.update(db, user, user_data)

# Delete
with get_db() as db:
    crud_user.remove(db, user_id)

from sqlalchemy import create_engine, URL
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from typing import AsyncGenerator

def create_engine_and_session(url: str | URL):
    try:
        # 数据库引擎
        engine = create_async_engine(url, pool_pre_ping=True)
    except Exception as e:
        print("❌ 数据库链接失败 {}", e)
        sys.exit()
    else:
        db_session = async_sessionmaker(
            bind=engine, autoflush=False, expire_on_commit=False
        )
        return engine, db_session


# 异步处理
async_engine, async_session = create_engine_and_session(settings.DB_URI_ASYNC)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """创建一个 SQLAlchemy 数据库会话-异步处理."""
    async with async_session() as session:
        yield session

ModelType = TypeVar("ModelType", bound=Base)
PrimaryKeyType = TypeVar("PrimaryKeyType", int, str, float)  # 限定主键的类型
PageDtoType = TypeVar("PageDtoType", bound=BaseModel)
DtoType = TypeVar("DtoType", bound=BaseModel)


class BaseCrud(Generic[ModelType, PrimaryKeyType, PageDtoType, DtoType]):
    """
    基础CRUD操作类
    """

    def __init__(self, model: Type[ModelType]):
        """
        数据库访问操作的基类对象(CRUD).
        **Parameters**
        * `model`: A SQLAlchemy model class
        """
        self.model = model

    async def get_all(
        self, sorting: Optional[str], db: AsyncSession
    ) -> List[ModelType] | None:
        """根据ID字符串列表获取对象列表

        :param sorting: 格式：name asc 或 name asc,age desc
        """
        query = select(self.model)
        if sorting:
            query = self.apply_sorting(query, sorting)

        result = await db.execute(query)
        items = result.scalars().all()
        return items

    async def get(self, id: PrimaryKeyType, db: AsyncSession) -> Optional[ModelType]:
        """根据主键获取一个对象"""
        query = select(self.model).filter(self.model.id == id)

        result = await db.execute(query)
        item = result.scalars().first()

        return item

    async def create(self, obj_in: DtoType, db: AsyncSession, **kwargs) -> bool:
        """创建对象,使用 kwargs 时可以扩展创建对象时的字段。

        :param obj_in: 对象输入数据
        :param kwargs: 扩展字段,如格式: is_deleted=0, is_active=1
        """
        try:
            if kwargs:
                instance = self.model(**obj_in.model_dump(), **kwargs)
            else:
                instance = self.model(**obj_in.model_dump())  # type: ignore

            db.add(instance)
            await db.commit()
            return True
        except SQLAlchemyError as e:
            print(e)
            await db.rollback()
            return False

from crud.customer import customer as customer_crud
from models.customer import Customer
from pydantic import BaseModel
from schemas.customer import CustomerDto, CustomerPageDto

async def test_list_customer():
    async with get_db() as db:

        print("get_list")
        totalCount, items = await customer_crud.get_list(
            CustomerPageDto(skipCount=0, maxResultCount=10, name="test"),
            db,
        )
        print(totalCount, items)
        for customer in customers:
            print(customer.name, customer.age)

        print("get_by_name")
        name = "test"
        customer = await customer_crud.get_by_name(
            name,
            db,
        )
        if customer:
            print(customer.name, customer.age)
        else:
            print(f"{name} not found")

        print("soft delete")
        result = await customer_crud.delete_byid(customer.id, db, is_deleted=1)
        print("操作结果：", result)

        print("soft delete_byids")
        result = await customer_crud.delete_byids(
            ["11122", "2C5F8672-2AA7-4B14-85AD-DF56F5BF7F1F"], db, is_deleted=1
        )
        print(f"Soft delete successful: {result}")

        print("update_by_column")
        result = await customer_crud.update_by_column(
            "id", customer.id, {"age": 30}, db
        )
        print("操作结果：", result)

        await db.close()