面试题：Python中利用SQLAlchemy实现复杂条件的分组查询

from sqlalchemy import create_engine, Column, Integer, Date, Float
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 创建数据库引擎
engine = create_engine('数据库连接字符串')

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 声明基类
Base = declarative_base()

# 定义orders表模型
class Orders(Base):
    __tablename__ = 'orders'
    order_id = Column(Integer, primary_key=True)
    customer_id = Column(Integer)
    order_date = Column(Date)
    total_amount = Column(Float)

# 查询每个客户在2023年下单的总金额，并且只返回总金额大于1000的客户及其总金额，结果按总金额降序排列
from sqlalchemy import func
result = session.query(Orders.customer_id, func.sum(Orders.total_amount).label('total_sum')) \
    .filter(Orders.order_date >= '2023-01-01', Orders.order_date < '2024-01-01') \
    .group_by(Orders.customer_id) \
    .having(func.sum(Orders.total_amount) > 1000) \
    .order_by(func.sum(Orders.total_amount).desc()) \
    .all()

for row in result:
    print(f"客户ID: {row.customer_id}, 总金额: {row.total_sum}")

session.close()

请将'数据库连接字符串'替换为实际的数据库连接信息，例如对于SQLite是'sqlite:///test.db'，对于MySQL是'mysql+pymysql://用户名:密码@主机:端口/数据库名' 。上述代码实现了题目需求，先定义表模型，然后使用SQLAlchemy的查询语法完成筛选、分组、过滤及排序操作。

面试题：Python中利用SQLAlchemy实现复杂条件的分组查询

知识考点

面试题答案