from sqlalchemy import create_engine, Column, Integer, String, Float, Date
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime, timedelta
# 创建数据库引擎
engine = create_engine('sqlite:///example.db')
# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
# 声明基类
Base = declarative_base()
# 用户表模型
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
city = Column(String)
# 订单表模型
class Order(Base):
__tablename__ = 'orders'
id = Column(Integer, primary_key=True)
user_id = Column(Integer)
amount = Column(Float)
order_date = Column(Date)
# 查找年龄大于30岁且在过去一个月内有订单金额大于1000的用户
one_month_ago = datetime.now().date() - timedelta(days=30)
result = session.query(User).join(Order, User.id == Order.user_id)\
.filter(User.age > 30, Order.amount > 1000, Order.order_date >= one_month_ago)\
.all()
for user in result:
print(user.name)
# 调试技巧:
1. **打印SQL语句**:
- 使用`print(str(query.statement.compile(compile_kwargs={"literal_binds": True})))`,这里`query`是构建的SQLAlchemy查询对象,通过打印SQL语句,可以检查SQL的语法以及条件是否正确,例如是否正确连接了表,过滤条件是否符合预期。
2. **检查数据类型**:
- 确认数据库表中列的数据类型与查询中使用的数据类型一致。比如日期类型的比较,确保`order_date`是`Date`类型,并且`one_month_ago`也是合适的日期类型。
- 可以在查询前,通过`print(User.age.type)`等方式检查列的数据类型。
3. **分步查询**:
- 先分别执行简单的查询,例如先查询年龄大于30岁的用户`users_over_30 = session.query(User).filter(User.age > 30).all()`,再查询过去一个月内订单金额大于1000的订单`orders_over_1000 = session.query(Order).filter(Order.amount > 1000, Order.order_date >= one_month_ago).all()`。
- 然后逐步组合查询,这样可以更容易定位问题出在哪个部分的查询逻辑上。
4. **日志记录**:
- 在SQLAlchemy中配置日志记录,例如:
```python
import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
- 这会记录SQLAlchemy执行的详细日志,包括连接数据库、执行SQL语句等信息,有助于分析查询执行过程中的问题。