SQLAlchemy ORM 增删改查

1from sqlalchemy import and_
2from sqlalchemy.orm import Session
3from api.models.user import User
4from db.base import get_db
5
6@router.get("/crud")
7async def crud(db: Session = Depends(get_db)):
8    # 查询全部
9    users = db.query(User).all()
10    if users:
11        user_id = users[0].username
12        print(user_id)
13
14    # 查询过滤
15    username = "admin"
16    user = db.query(User).filter(and_(User.username == username)).first()
17    if user:
18        user_id = user.id
19        print(user_id)
20
21    # 新增
22    new_user = User(username='Zailiang', name="孙")
23    db.add(new_user)
24    db.commit()
25    db.refresh(new_user)
26    print(new_user)
27
28    # 删除
29    user = db.query(User).filter(and_(User.id == 3)).first()
30    db.delete(user)
31    db.commit()
32
33    # 修改
34    user = db.query(User).filter(and_(User.id == 13)).first()
35    user.name = "ZL"
36    user.username = "SUN"
37    db.commit()
38    db.refresh(user)

SQLAlchemy 原生查询

1async def search_article_by_id(request: Request, db: Session = Depends(get_db)):
2    """根据id查询文章"""
3    form = await request.form()
4    id = form.get('id')
5    info = db.execute(text("SELECT * FROM t_article_info WHERE id = :id"), {"id": id})
6    # 使用属性访问
7    one = info.fetchone()
8    article_title = one.article_title
9    # 使用索引访问
10    one = info.mappings().fetchone()
11    pk = one['pk']

通过属性名访问：

假设 label 是数据库中的一个列名，你可以直接通过 Row 对象的属性来访问

映射为字典风格访问：

如果你更喜欢字典风格的访问方式，你可以通过 SQLAlchemy 的 Result 对象调用 .mappings() 方法，将行结果映射为字典对象。这将允许你使用 one['label'] 的方式访问字段值。

异步 SQL 查询

1@router.get("/users")
2async def get_users(db: AsyncSession = Depends(get_db)):
3    # orm 查询单个
4    stmt = select(User).filter(User.id == 1)
5    execute_result = await db.execute(stmt)
6    user = execute_result.scalars().first()
7    print(user.username)
8    # orm 查询多个
9    stmt = select(User)
10    execute_result = await db.execute(stmt)
11    users = execute_result.scalars().all()
12    for user in users:
13        print(user.username)
14    # 原生查询
15    execute = await db.execute(text("SELECT * FROM user"))
16    fetchall = execute.fetchall()
17    print(fetchall)

SQLAlchemy 结果用字段名称获取

1from sqlalchemy import text
2
3sql = text("SELECT text, tokens FROM milvus_case_tokens")
4info = await db.execute(sql)
5# 使用 mappings() 将每一行转换为 dict-like 对象
6rows = info.mappings().all()
7for row in rows:
8    print(row["tokens"])  # 使用字段名访问

mappings() 方法会将每一行转换成一个字典格式（RowMapping 对象），你就可以通过字段名来访问数据了