实现

创建目录

在 main.py 同级目录，创建 crud（或 repository、dao） 目录，存放各个表的增删改查操作文件

创建 user 操作文件

在 crud 目录，创建 user.py 文件，内容如下：

from sqlalchemy import delete, update, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from models.user import User


async def save(session:AsyncSession ,user: User):
    try:
        session.add(user)
        # 成功执行才提交事务
        await session.commit()
    except Exception as e:
        # 如果报错就回滚
        await session.rollback()
        raise e


async def delete_user(session:AsyncSession ,id:int):
    """根据id删除数据"""
    try:
        stmt = delete(User).where(User.id == id)
        await session.execute(stmt)
        # 成功执行才提交事务
        await session.commit()
    except Exception as e:
        # 如果报错就回滚
        await session.rollback()
        raise e


async def update_user(session:AsyncSession ,id:int,password:str):
    """根据id更新数据"""
    try:
        stmt = update(User).where(User.id == id).values(password=password)
        await session.execute(stmt)
        # 成功执行才提交事务
        await session.commit()
    except Exception as e:
        # 如果报错就回滚
        await session.rollback()
        raise e


async def query_all(session:AsyncSession ,):
    """查询所有数据"""
    stmt = select(User)
    res = await session.execute(stmt)
    user_list = res.scalars().all()  # scalars获取模型对象
    return user_list


async def query_by_id(session:AsyncSession ,id:int):
    """查询一条数据"""
    stmt = select(User).where(User.id == id)
    res = await session.execute(stmt)
    # 返回一条数据，或没有数据
    user = res.scalar_one_or_none()
    return user


async def query_sql(session:AsyncSession ,sql:str,params:tuple):
    """查询原生SQL，执行复杂SQL，否则不建议使用，因为不能跨数据库"""
    sql_t = text(sql)
    result = await session.execute(sql_t, params)
    rows = result.mappings().all()
    return rows