使用 SqlAlchemy 的自定义类型,为什么使用 from_statement 时 process_result_value 的结果没有传输?
Using custom type with SqlAlchemy, why are the results of process_result_value not transferred when using from_statement?
我正在使用自定义类型表单 this answer, which is quite similar to the one recommended in the SqlAlchemy documentation,让我在这里复制它作为一个独立的问题:
from sqlalchemy import types
from datetime import datetime, timezone
class UTCDateTime(types.TypeDecorator):
impl = types.DateTime
def process_bind_param(self, value, engine):
if value is None:
return
if value.utcoffset() is None:
raise ValueError(
'Got naive datetime while timezone-aware is expected'
)
return value.astimezone(timezone.utc)
def process_result_value(self, value, engine):
if value is not None:
print('passed here')
return value.replace(tzinfo=timezone.utc)
如果我这样查询:
session.query(Table).filter_by(...).first()
..一切按预期工作,方法 process_result_value
运行并且我的日期时间对象带有 UTC tzinfo。
但是,如果我这样查询:
db.query(User).from_statement(text("SELECT * from table where ...")).first()
.. 然后方法 process_result_value
运行(我知道是因为我在控制台中看到打印语句),但返回值没有结果中的 tzinfo。
我正在使用 SqlAlchemy 1.3.23。
我的问题是:这是预期的行为吗?有解决办法吗?
我的问题的解决方法是不依赖文本查询中 id
的文本替换 *
。
它没有记录在 Query.from_statement
上,但我发现我可以将它用于 returns 仅主键并让 SqlAlchemy 解析剩余字段的查询。
这解决了问题,因为 SqlAlchemy 不会直接根据 SQL 查询的结果进行文本映射,而是正确地进行映射,代价是对每个结果 ID 进行一次额外查询。
我正在使用自定义类型表单 this answer, which is quite similar to the one recommended in the SqlAlchemy documentation,让我在这里复制它作为一个独立的问题:
from sqlalchemy import types
from datetime import datetime, timezone
class UTCDateTime(types.TypeDecorator):
impl = types.DateTime
def process_bind_param(self, value, engine):
if value is None:
return
if value.utcoffset() is None:
raise ValueError(
'Got naive datetime while timezone-aware is expected'
)
return value.astimezone(timezone.utc)
def process_result_value(self, value, engine):
if value is not None:
print('passed here')
return value.replace(tzinfo=timezone.utc)
如果我这样查询:
session.query(Table).filter_by(...).first()
..一切按预期工作,方法 process_result_value
运行并且我的日期时间对象带有 UTC tzinfo。
但是,如果我这样查询:
db.query(User).from_statement(text("SELECT * from table where ...")).first()
.. 然后方法 process_result_value
运行(我知道是因为我在控制台中看到打印语句),但返回值没有结果中的 tzinfo。
我正在使用 SqlAlchemy 1.3.23。
我的问题是:这是预期的行为吗?有解决办法吗?
我的问题的解决方法是不依赖文本查询中 id
的文本替换 *
。
它没有记录在 Query.from_statement
上,但我发现我可以将它用于 returns 仅主键并让 SqlAlchemy 解析剩余字段的查询。
这解决了问题,因为 SqlAlchemy 不会直接根据 SQL 查询的结果进行文本映射,而是正确地进行映射,代价是对每个结果 ID 进行一次额外查询。