SQLAlchemy异步模式返回UTC时区时间的异常问题

环境：PostgreSQL 15，SQLAlchemy 2.0.7

执行set local time zone 'Europe/Berlin';设置时区后，异步模式下SQLAlchemy返回的datetime始终为UTC时区时间，但同步模式及直接使用psycopg2时，能正常返回柏林时区（+02:00）的时间。

测试代码 #

import asyncio
import logging
import sys
from datetime import datetime

import psycopg2
import psycopg2.extras
from sqlalchemy import (
    Result,
    create_engine,
    text,
)
from sqlalchemy.ext.asyncio import (
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import sessionmaker

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logging.getLogger("asyncio").setLevel(logging.INFO)


async def async_test_tz(dsn) -> datetime:
    async_engine = create_async_engine(dsn, echo=False)
    async_session = async_sessionmaker(async_engine, expire_on_commit=False)
    async with async_session() as session:
        async with session.begin():
            timezone_query = "set local time zone 'Europe/Berlin'"
            await session.execute(text(timezone_query))
            sql_query = text("select now()::timestamptz as result")
            results: Result = await session.execute(sql_query)
            return results.fetchone()._asdict()["result"]  # type: ignore


async def async_test_tz_with_engine(dsn) -> datetime:
    async_engine = create_async_engine(dsn, echo=False)
    async with async_engine.begin() as session:
        timezone_query = "set local time zone 'Europe/Berlin'"
        await session.execute(text(timezone_query))
        sql_query = text("select now()::timestamptz as result")
        results: Result = await session.execute(sql_query)
        return results.fetchone()._asdict()["result"]  # type: ignore


async def async_test_tz_with_engine_run_sync(dsn) -> datetime:
    def set_timezone(conn):
        timezone_query = "set local time zone 'Europe/Berlin'"
        conn.execute(text(timezone_query))

    def get_date(conn) -> datetime:
        sql_query = text("select now()::timestamptz as result")
        results: Result = conn.execute(sql_query)
        return results.fetchone()._asdict()["result"]  # type: ignore

    async_engine = create_async_engine(dsn, echo=False)
    async with async_engine.begin() as session:
        await session.run_sync(set_timezone)
        return await session.run_sync(get_date)


def psycopg2_sync_test_tz(dsn) -> datetime:
    with psycopg2.connect(dsn) as connection:
        with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
            timezone_query = "set local time zone 'Europe/Berlin'"
            cursor.execute(timezone_query)
            sql_query = "select now()::timestamptz as result"
            cursor.execute(sql_query)
            results = cursor.fetchone()
            return results["result"]  # type: ignore


def sync_test_tz(dsn) -> datetime:
    sync_engine = create_engine(dsn, echo=False)
    sync_session = sessionmaker(sync_engine, expire_on_commit=False)
    with sync_session() as session:
        timezone_query = "set local time zone 'Europe/Berlin'"
        session.execute(text(timezone_query))
        sql_query = text("select now()::timestamptz as result")
        results: Result = session.execute(sql_query)
        return results.fetchone()._asdict()["result"]  # type: ignore


def sync_test_tz_with_engine(dsn) -> datetime:
    sync_engine = create_engine(dsn, echo=False)
    with sync_engine.begin() as session:
        timezone_query = "set local time zone 'Europe/Berlin'"
        session.execute(text(timezone_query))
        sql_query = text("select now()::timestamptz as result")
        results: Result = session.execute(sql_query)
        return results.fetchone()._asdict()["result"]  # type: ignore


def start():
    async_dsn = sys.argv[1]
    sync_dsn = sys.argv[2]

    loop = asyncio.get_event_loop()

    res = loop.run_until_complete(async_test_tz(async_dsn))
    logger.info("Async sqlalchemy session maker: %s", res)

    res = loop.run_until_complete(async_test_tz_with_engine(async_dsn))
    logger.info("Async sqlalchemy engine: %s", res)

    res = loop.run_until_complete(async_test_tz_with_engine_run_sync(async_dsn))
    logger.info("Async sqlalchemy engine with run_sync: %s", res)

    res = sync_test_tz(sync_dsn)
    logger.info("Sync sqlalchemy session maker: %s", res)

    res = sync_test_tz_with_engine(sync_dsn)
    logger.info("Sync sqlalchemy engine: %s", res)

    res = psycopg2_sync_test_tz(sync_dsn)
    logger.info("psycopg2: %s", res)


if __name__ == "__main__":
    start()

测试结果 #

INFO:main: Async sqlalchemy session maker: 2023-06-16 22:07:44.378642+00:00
INFO:main: Async sqlalchemy engine: 2023-06-16 22:07:44.411493+00:00
INFO:main: Async sqlalchemy engine with run_sync: 2023-06-16 22:07:44.445468+00:00
INFO:main: Sync sqlalchemy session maker: 2023-06-17 00:07:44.456777+02:00
INFO:main: Sync sqlalchemy engine: 2023-06-17 00:07:44.468090+02:00
INFO:main: psycopg2: 2023-06-17 00:07:44.473602+02:00

所有异步模式下的SQLAlchemy测试均返回*UTC（+00:00）时间，而同步模式SQLAlchemy及直接使用psycopg2的测试，均返回正确的柏林时区（+02:00）*时间。

内容的提问来源于stack exchange，提问作者polymetr