Существует ли функция FROM_UNIXTIME (), не зависящая от базы данных SqlAlchemy? - PullRequest
0 голосов
/ 08 февраля 2019

В настоящее время у меня есть запрос, похожий на приведенный ниже, во флаконе sqlalchemy:

from sqlalchemy.sql import func

models = (
  Model.query
  .join(ModelTwo)
  .filter(Model.finish_time >= func.from_unixtime(ModelTwo.start_date))
  .all()
)

Это прекрасно работает с MySQL, который я запускаю в производственной среде, однако, когда я запускаю тесты для метода с использованием встроенной памятиСбой базы данных SqlLite, потому что from_unixtime не является функцией SqlLite.

Помимо выполнения тестов на той же базе данных, что и производство, насколько это возможно, проблема и тот факт, что у меня есть два разных способа представления данных вбаза данных, есть ли в SqlAlchemy метод, независимый от базы данных, для обработки преобразования дат в метки времени Unix и наоборот?

1 Ответ

0 голосов
/ 08 февраля 2019

Для всех, кто интересуется этим, я нашел способ создания пользовательских функций в SqlAlchemy на основе используемого диалекта SQL.Таким образом, ниже достигается то, что мне нужно:

from sqlalchemy.sql import expression
from sqlalchemy.ext.compiler import compiles


class convert_timestamp_to_date(expression.FunctionElement):
    name = 'convert_timestamp_to_date'


@compiles(convert_timestamp_to_date)
def mysql_convert_timestamp_to_date(element, compiler, **kwargs):
    return 'from_unixtime({})'.format(compiler.process(element.clauses))


@compiles(convert_timestamp_to_date, 'sqlite')
def sqlite_convert_timestamp_to_date(element, compiler, **kwargs):
    return 'datetime({}, "unixepoch")'.format(compiler.process(element.clauses))

Вышеприведенный запрос теперь можно переписать так:

models = (
  Model.query
  .join(ModelTwo)
  .filter(Model.finish_time >= convert_timestamp_to_date(ModelTwo.start_date))
  .all()
)
...