SQLAlchemy ORM查询Oracle时区Timestamp报错及列类型转换方案
解决方案
方案一:直接修复ORA-01805时区错误
ORA-01805本质是客户端与Oracle服务器的时区配置不兼容,可通过以下步骤解决:
统一客户端与服务器时区
在创建SQLAlchemy引擎时,指定与Oracle服务器一致的时区:from sqlalchemy import create_engine import os # 设置环境变量,替换为你的Oracle服务器时区(如Asia/Shanghai) os.environ["TZ"] = "Asia/Shanghai" # 创建引擎时传入时区参数 engine = create_engine( "oracle+oracledb://user:password@host:port/service_name", connect_args={"timezone": "Asia/Shanghai"} )调整模型字段的时区处理
针对带时区的TIMESTAMP字段,在模型定义中明确服务器端默认值与更新逻辑:from sqlalchemy import TIMESTAMP, text from dataclasses import field # 修改date_created和date_modified字段定义 date_created: datetime = field( metadata={"sa": Column(TIMESTAMP(True), nullable=False, server_default=text("CURRENT_TIMESTAMP"))} ) date_modified: datetime = field( metadata={"sa": Column(TIMESTAMP(True), nullable=False, server_default=text("CURRENT_TIMESTAMP"), onupdate=text("CURRENT_TIMESTAMP"))} )切换python-oracledb厚模式(可选)
若为时区文件版本不匹配问题,切换到厚模式并确保安装对应版本的Oracle客户端:import oracledb oracledb.init_oracle_client() # 需提前配置Oracle客户端环境变量
方案二:自动将Timestamp字段转为字符串(无需逐个指定列)
通过SQLAlchemy的事件监听或混合属性,实现查询时自动格式化指定Timestamp字段,无需手动编写列转换逻辑:
方法1:SQLAlchemy事件监听拦截查询
在查询编译前自动替换目标字段为func.to_char转换后的结果:
from sqlalchemy import func, event from sqlalchemy.orm import Query # 定义需要转换的Timestamp字段名 TARGET_FIELDS = ["date_created", "date_modified"] @event.listens_for(Query, "before_compile", retval=True) def auto_convert_timestamp(query): for desc in query.column_descriptions: if desc["type"] is Department: # 替换原有字段为格式化后的字符串字段 new_entities = [] for col in desc["entity"].__mapper__.columns: if col.key in TARGET_FIELDS: converted_col = func.to_char(col, 'YYYY-MM-DD HH24:MI:SS TZR').label(col.key) new_entities.append(converted_col) else: new_entities.append(col) query = query.with_entities(*new_entities) return query
执行db.query(Department).all()时,date_created和date_modified会自动转为格式化字符串,需同步修改DepartmentResponse中对应字段类型为str。
方法2:使用混合属性覆盖查询逻辑
在Department模型中定义混合属性,同时支持内存计算与SQL查询转换:
from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy import func @mapper_registry.mapped @dataclass class Department: # 原有字段定义保持不变,新增混合属性 @hybrid_property def date_created_str(self): return self.date_created.strftime('%Y-%m-%d %H:%M:%S %Z') if self.date_created else None @date_created_str.expression def date_created_str(cls): return func.to_char(cls.date_created, 'YYYY-MM-DD HH24:MI:SS TZR') @hybrid_property def date_modified_str(self): return self.date_modified.strftime('%Y-%m-%d %H:%M:%S %Z') if self.date_modified else None @date_modified_str.expression def date_modified_str(cls): return func.to_char(cls.date_modified, 'YYYY-MM-DD HH24:MI:SS TZR')
修改DepartmentResponse模型使用date_created_str和date_modified_str字段,查询时可直接调用db.query(Department).options(load_only("id", "label", "date_created_str", "date_modified_str")).all()避免冗余数据加载。
内容的提问来源于stack exchange,提问作者Isaac
相关产品推荐
相关产品推荐

