sqlAlchemy 数据库查询

OrangBus大约 2 分钟

安装依赖

uv add sqlalchemy psycopg2-binary

数据库连接

from sqlalchemy.orm import declarative_base, sessionmaker, relationship, joinedload
from sqlalchemy import create_engine, Column, MetaData, Integer, String, DateTime,ForeignKey

engine = create_engine("postgresql://postgres:admin666@home.cc:5433/study?client_encoding=utf8",echo=True)
Base = declarative_base()

sesstion = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db = sesstion()

创建表


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True,autoincrement=True,comment="用户Id")
    name = Column(String(50),comment="用户名称")
    password = Column(String(20),comment="密码")
    phone = Column(String(15),comment="手机号")
    created_at = Column(DateTime,comment="创建时间",auto_now_add=True)

    # 一个用户有多个订单:
    orders = relationship("Order",back_populates="user")

class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True,autoincrement=True)
    name = Column(String(50),comment="订单名称")
    user_id = Column(Integer,ForeignKey("users.id"),comment="user:id")
    created_at = Column(DateTime,auto_now_add=True)

    # 一个订单对应一个用户
    user = relationship("User",back_populates="orders",uselist=False)
    # 一个订单对应一个本书
    book = relationship("Book",back_populates="order",uselist=False)

class Book(Base):
    __tablename__ = "books"
    id = Column(Integer, primary_key=True,autoincrement=True)
    name = Column(String(50),comment="文章名称")
    order_id = Column(Integer,ForeignKey("orders.id"),comment="order:id")
    created_at = Column(DateTime,auto_now_add=True)
  # 一个订单对应一本书：一对一
    order = relationship("Order",back_populates="book",uselist=False)


Base.metadata.create_all(engine)
print("数据表创建成功")

ORM 关系

增删改查

import datetime

def user_add():
    u = User(name="admin",password="admin666",phone="12345678901",created_at=datetime.datetime.now())
    db.add(u)
    db.commit()
    db.refresh(u)
    print("user id:",u.id)

def user_insert():
    user_list = []
    for i in range(10):
        user = User(name="admin"+str(i),password="admin666",phone="12345678901",created_at=datetime.datetime.now())
        userList.append(user)
    db.add_all(user_list)
    db.commit()

# user_add()
user_insert()

def get_one():
    user = db.query(User).filter(User.id==1).first()
    print(user.id)
    print(user.name)

def get_list():
    list = db.query(User).offset(0).limit(10)
    for u in list:
        print(u.id)

def user_count():
    count = db.query(User).filter(User.id > 10).count()
    print(count)
# get_one()
# get_list()
user_count()

user = db.query(User).filter(User.id==5).first()
if user.id > 0:
    user.name = "update name"
    db.commit()

# db.query(User).where(User.id == 1 ).delete()
db.query(User).where(User.id.in_([2,3])).delete()
db.commit()

分页查询

from sqlalchemy import func,select

def pagination(name:str,page:int,page_size:int):
    stmt = select(User)
    # 动态拼接条件
    filters = []
    if name:
        filters.append(User.name.like(f"%{name}%"))

    if filters:
        stmt = stmt.where(*filters)

    stmt = stmt.order_by(User.id.desc())
    # 分页
    page_query = stmt.offset((page-1)*page_size).limit(page_size)
    data = db.scalars(page_query).all()
    # 总数
    count_stmt = select(func.count(User.id)).where(*filters)
    total = db.scalar(count_stmt)

    return {
        "list": data,
        "total": total,
        "page": page,
        "page_size": page_size,
        "total_pages": (total + page_size - 1)
    }
list = pagination("admin",1,2)
for user in list["list"]:
    print(user)

关联查询

from sqlalchemy.orm import joinedload
# 一个人对应多个订单
user_with_order = db.query(User).options(joinedload(User.orders)).filter(User.id==1).first()
user_with_order.orders # 这里是一个列表

data = db.query(Order).options(joinedload(Order.book)).filter(Order.id==1).first()
print(data.book) # 这里是一个字典

关联统计

from sqlalchemy import select, func, coalesce

stmt = select(
    User.phone,
    User.name,
    func.count(Order.id).label("order_cnt"),
    coalesce(func.sum(Order.amount), 0).label("total_amount")
).join(
    Order, User.id == Order.user_id, isouter=True  # left join
).group_by(
    User.id, User.phone, User.name
).order_by(
    func.count(Order.id).desc()
)

res = db.execute(stmt).mappings().all()
for row in res:
    print(row.phone, row.name, row.order_cnt, row.total_amount)

分组统计

select phone,count("phone") as phone_total from users 
group by "phone" 
order by phone_total desc

stmt = select(
    User.phone,
    func.count(User.phone).label("phone_total")
).group_by(
    User.phone
).order_by(
    func.count(User.phone).desc()
)
group_list = db.execute(stmt).all()
for item in group_list:
    print(item)