如何在FastAPI中使用SQLAlchemy实现指定SQL查询逻辑

我正在把原有Python程序迁移到FastAPI/SQLAlchemy，纠结是复用现有SQL还是用SQLAlchemy的过滤功能。自己写过滤代码时遇到两个问题：一是arr_utc字段的30分钟时间偏移转换报错，二是OR条件不生效。我是SQLAlchemy新手，下面是我的Flights模型、尝试的过滤代码和之前能用的原生SQL，求帮忙解决。

模型定义 #

class Flights(Base):
    __tablename__ = "flights"

    id = Column(Integer, primary_key=True, index=True)
    timestamp = Column(TIMESTAMP)
    flight = Column(String)
    dept_city = Column(String)
    arr_city = Column(String)
    dept_utc = Column(String)
    arr_utc = Column(String)

尝试的过滤代码 #

now = datetime.now()

flight = db.query(Flights)

flight = flight.filter(datetime(Flights.arr_utc, '-30 minutes') >= now,
   Flights.dept_utc <= now,
   Flights.act_in == 'None' or Flights.act_in == Null,
   Flights.invalid=='')

原有可用原生SQL语句 #

dte = datetime.now()

c.execute("SELECT * from flights WHERE ? \
BETWEEN datetime(dept_utc,'-30 minutes') AND datetime(arr_utc,'+30 minutes') \
AND ((act_in = 'None') OR (act_in is null)) \
AND (invalid is NULL) \
ORDER BY timestamp DESC \
LIMIT 1;", [dte])

问题解决与优化方案 #

1. 时间偏移转换错误修复

你直接用Python的datetime()处理SQLAlchemy字段是错误的——Flights.arr_utc是ORM字段对象，不是原生Python字符串。要实现SQL里的datetime(arr_utc, '+30 minutes')逻辑，必须用SQLAlchemy的func调用数据库内置的datetime函数：

from sqlalchemy import func

# 对应SQL中的datetime(dept_utc,'-30 minutes')
dept_utc_minus_30 = func.datetime(Flights.dept_utc, '-30 minutes')
# 对应SQL中的datetime(arr_utc,'+30 minutes')
arr_utc_plus_30 = func.datetime(Flights.arr_utc, '+30 minutes')

同时原SQL的时间判断是now BETWEEN 两个偏移时间，过滤条件要对应调整：

now.between(dept_utc_minus_30, arr_utc_plus_30)

2. OR条件不生效修复

Python的or是本地逻辑运算符，不能直接用于SQLAlchemy的过滤条件，必须用SQLAlchemy提供的or_()函数；另外判断字段为NULL不能用== None，要使用is_(None)方法：

from sqlalchemy import or_

# 替换原有的act_in判断条件
or_(Flights.act_in == 'None', Flights.act_in.is_(None))

同时你写的Flights.invalid==''和原SQL的invalid is NULL不符，要改成Flights.invalid.is_(None)。

3. 修正后的完整过滤代码

from datetime import datetime
from sqlalchemy import func, or_

now = datetime.now()
flight = db.query(Flights).filter(
    now.between(func.datetime(Flights.dept_utc, '-30 minutes'), func.datetime(Flights.arr_utc, '+30 minutes')),
    or_(Flights.act_in == 'None', Flights.act_in.is_(None)),
    Flights.invalid.is_(None)
).order_by(Flights.timestamp.desc()).limit(1)

4. 复用SQL还是用ORM过滤？

• 短期过渡：如果原有SQL逻辑复杂，可直接用SQLAlchemy的text()执行原生SQL，成本更低：

  from sqlalchemy import text
  
  dte = datetime.now()
  result = db.query(Flights).from_statement(text("""
      SELECT * from flights WHERE :dte 
      BETWEEN datetime(dept_utc,'-30 minutes') AND datetime(arr_utc,'+30 minutes') 
      AND ((act_in = 'None') OR (act_in is null)) 
      AND (invalid is NULL) 
      ORDER BY timestamp DESC 
      LIMIT 1;
  """)).params(dte=dte).first()
  

• 长期优化：建议逐步迁移到ORM过滤，优势是类型安全、代码可读性高、和FastAPI的Pydantic模型兼容性更好。

内容的提问来源于stack exchange，提问作者user1134287