FastAPI教程-集成SQLAlchemy：封装表增删改查的操作文件-马育民老师

# 实现

### 创建目录

在 `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
```

原文出处：/show_1GW3sRsdJ1bK.html