sqlalchemy date как строка в raw sql

Рассмотрим следующую таблицу:

class Employee(Base): __tablename__ = "t_employee" id = Column(Integer(20), Sequence('%s_id_seq' % __tablename__), primary_key=True) first_name = Column(String(30)) last_name = Column(String(30)) email = Column(String(50)) start_date = Column(Date, default=datetime.now) end_date = Column(Date) 

Как выбрать, используя строки вместо дат в сыром sql в sqlalchemy? Следующие действия выполняются в mysql, но не в Oracle:

 session.query(Employee).\ filter("end_date IS NULL OR end_date>='%s'" % datetime.now()).all() 

Лучший сценарий был бы, если бы я мог использовать строку или дату (взаимозаменяемо) при работе с столбцами Date или DateTime (я пробовал TypeDecorator безрезультатно)

Обратите внимание, что вопрос относится к raw sql (я знаю, что это можно сделать с использованием предикатов) …

Не интерполируйте значения в SQL с помощью форматирования строк. В случае объекта datetime формат строки по умолчанию работает для MySQL, но это просто удача и удача.

В этом случае не используйте raw SQL и SQLAlchemy переведите объект datetime на то, что база данных базы данных понимает для вас:

 from sqlalchemy import or_ session.query(Employee).filter( or_(Employee.end_date == None, Employee.end_date >= datetime.now()) ).all() 

Даже при использовании необработанного SQL я бы sqlalchemy сгенерировал SQL и использовал параметры привязки:

 from sqlalchemy.sql.expression import bindparam, column from sqlalchemy.types import DateTime from sqlalchemy import or_ dtnow = bindparam('dtnow', datetime.now(), DateTime) end_date = column('enddate', DateTime) session.query(Employee).\ filter(or_(end_date == None, end_date >= dtnow)).all() 

Это выражение фильтра превращается в правильно экранированный SQL для вашего бэкэнда базы данных, независимо от того, что может быть. Когда бэкэнд не установлен, выражение становится:

 >>> str(or_(end_date == None, end_date >= dtnow)) 'enddate IS NULL OR enddate >= :dtnow' 

и значение datetime.now() будет передано в качестве параметра SQL во внутренний буфер базы данных при выполнении.

Последнее средство – использовать тип text() :

 from sqlalchemy.sql.expression import bindparam, text dtnow = bindparam('dtnow', datetime.now(), DateTime) session.query(Employee).\ filter(text('end_date is NULL or end_date >= :dtnow', bindparams=[dtnow])).all() 

В противном случае я бы вообще не смешивал исходный SQL и ORM SQLAlchemy. Использовать необработанный SQL непосредственно с подключением к базе данных:

 conn = session.connection() conn.execute(text('SELECT * FROM t_employee WHERE end_date IS NULL OR end_date>=:dtnow'), dtnow=datetime.now())