在 SQLAlchemy 中截断小时/天/周/月/年

Question

有没有办法在适用于所有 DBMS（尤其是 Postgres 和 SQLite）的 SQLAlchemy 中将日期时间截断为小时/天/周/月/年？

Answer 1

不是开箱即用，但你可以制作一个（我想这与this有关）：

from sqlalchemy import func
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import ColumnElement
from sqlalchemy.types import DateTime

class Trunc(ColumnElement):

    type = DateTime()

    def __init__(self, precision, expr):
        self.precision = precision
        self.expr = expr

    @property
    def _from_objects(self):
        return self.expr._from_objects

Postgresql 的编译器很简单：

@compiles(Trunc, 'postgresql')
def compile_trunc_postgresql(element, compiler, **kw):
    return compiler.process(func.date_trunc(element.precision, element.expr))

SQLite 版本更复杂，因为没有一站式的截断解决方案：

_modifiers = {
    'year': ('start of year',),
    'month': ('start of month',),
    # This does not account for locale specific first day of week. 1 day
    # is added so that the 1st day of week won't truncate to previous week.
    # Replace 'weekday 0' with 'weekday 1', if you'd like first day of
    # week to be Monday (in accordance with ISO 8601)
    'week': ('1 day', 'weekday 0', '-7 days', 'start of day'),
    'day': ('start of day',),
}

@compiles(Trunc, 'sqlite')
def compile_trunc_sqlite(element, compiler, **kw):
    precision = element.precision
    expr = element.expr
    modifiers = _modifiers.get(precision)

    if modifiers:
        return compiler.process(func.datetime(expr, *modifiers))

    elif precision == 'hour':
        return compiler.process(func.datetime(
            expr,
            func.strftime('-%M minutes', expr),
            func.strftime('-%f seconds', expr)))

    elif precision == 'minute':
        return compiler.process(func.datetime(
            expr, func.strftime('-%f seconds', expr)))

    elif precision == 'second':
        return compiler.process(func.datetime(
            expr,
            func.strftime('-%f seconds', expr),
            func.strftime('%S seconds', expr)))

SQLite 版本不支持 Postgresql 中所有可用的精度修饰符，例如“季度”，但应该相当有用。用法：

In [16]: for p in ['year', 'month', 'week', 'day', 'hour', 'minute', 'second']:
    ...:     print(engine.execute(select([Trunc(p, func.current_timestamp())])).scalar())
    ...:     
2018-01-01 00:00:00
2018-08-01 00:00:00
2018-07-29 00:00:00
2018-08-03 00:00:00
2018-08-03 06:00:00
2018-08-03 06:18:00
2018-08-03 06:18:18