Postgres 函数作为 SQLAlchemy 中的列

Question

我在 Postgesql 数据库中有两个函数：

id_generator，从时间戳
timestamp_from_id(id)，将 id 反转为 timestamp

假设这是我的模型：

class DatModel(object):
    id = Column(
        BigInteger,
        primary_key=True,
        server_default=text('id_generator()')
    )
    name = Column(String(50), index=True)

我要做的是通过timestamp_from_id(id)函数生成的timestamp进行查询，例如：

dbsession.query(DatModel).filter(DatModel.timestamp > datetime.now()).all()

或

obj = dbsession.query(DatModel).first()
created = obj.timestamp

我的问题是：

如何基于postgres函数创建虚拟列？

提前致谢

编辑：

不满意的方法 #1

作为Roman Dryndik possible solution is to use func。这是工作方法，但在检索时间戳时几乎没有问题：

timestamp = session.execute(func.timestamp_from_id(obj.id)).fetchone()

我更愿意这样做：

timestamp = obj.timestamp

不满意的方法 #2

第二种可能的解决方案是使用Compilation Extension

from datetime import datetime

from sqlalchemy import DateTime
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.functions import FunctionElement

class Timestamp(FunctionElement):
    type = DateTime()
    name = 'timestamp'

@compiles(Timestamp)
def default_timestamp(element, compiler, **kw):
    return compiler.visit_function(element)

@compiles(Timestamp, 'postgresql')
def pg_created(element, compiler, **kw):
    arg1 = list(element.clauses)[0]
    return 'timestamp_from_id({})'.format(compiler.process(arg1))


# filter example
obj = session.query(DatModel).filter(Timestamp < datetime.now()).first()

# get timestamp attribute example
timestamp = session.execute(Timestamp(obj.id)).fetchone()

Answer 1

如果我正确理解您的问题，您可以使用 func.

调用存储过程 timestamp_from_id

所以你会得到这样的东西：

from sqlalchemy import func

# some code here

dbsession.query(DatModel).filter(func.timestamp_from_id(DataModel.id) > datetime.now()).all()

# some code here

没有尝试代码。也许你也应该做 func.timestamp_from_id(DataModel.id).scalar()。

您可以找到更多信息 here and here。

Answer 2

请阅读 SQL 表达式作为映射属性 文档的 Using column_property 部分。

使用它你应该可以做到：

class DatModel(Base):
    id = Column(
        Integer,
        primary_key=True,
        server_default=text('id_generator()')
    )
    name = Column(String(50), index=True)

    timestamp = column_property(func.timestamp_from_id(id))

timestamp 也将包含在您的查询中，您可以在过滤 and/or 排序表达式中使用它：

q = (session.query(DatModel)
     .filter(DatModel.timestamp > datetime.now())
     .order_by(DatModel.timestamp.desc())
    )

Answer 3

解决方案是使用hybrid_property和expression (docs)

class DatModel(Base)::
    id = Column(
        BigInteger,
        primary_key=True,
        server_default=text('id_generator()')
    )

    @hybrid_property
    def timestamp(self):
        return object_session(self). \
            scalar(select([func.timestamp_from_id(self.id)]))

    @created.expression
    def timestamp(self):
        return func.timestamp_from_id(self.id)

用法示例：

obj = DatModel()
with transaction.manager:
    session.add(obj)
obj = session.query(DatModel).one()

assert type(obj.timestamp) == datetime.datetime
assert obj.timestamp.date() == datetime.date.today()


objs = session.query(DatModel) \
    .filter(DatModel.timestamp < datetime.datetime.now())
assert objs.count() == 1

objs = session.query(Gateway) \
    .filter(DatModel.timestamp > datetime.datetime.now())
assert objs.count() == 0

Answer 4

您可以使用 column_property with a literal_column 实现此目的。例如：

class DatModel(object):
    id = Column(
        BigInteger,
        primary_key=True,
        server_default=text('id_generator()')
    )
    name = Column(String(50), index=True)
    timestamp = column_property(literal_column("timestamp_from_id(datmodel.id)", type_=DateTime).label('timestamp'))

Postgres 函数作为 SQLAlchemy 中的列

Postgres function as column in SQLAlchemy

python

postgresql

sqlalchemy

不满意的方法 #1

不满意的方法 #2