使用SQLAlchemy查询ClickHouse时的日期类型匹配问题求助

问题场景 #

通过SQLAlchemy向ClickHouse发送查询时，因DateTime类型处理不兼容触发错误，相关代码及报错信息如下：

查询函数代码 #

def get_active_contract(session: DBSession, **kwargs):
    date = kwargs['date'] if '_date' in kwargs else datetime.today()
    subquery = session.query(
        Contract_State_History.id.label("id"), Contract_State_History.date.label("date"), Contract_State_History.state.label("state"),
        func.rank().over(
            order_by=Contract_State_History.date.desc(),
            partition_by=Contract_State_History.id
        ).label("rnk")
    ).filter(Contract_State_History.date<=date).subquery()
    query = session.query(subquery.c.id.label("contract_id"), 
                        Contract.code.label("contract_code"), 
                        Contract.name.label("contract_name"), 
                        subquery.c.state.label("contract_state"),
                        Contract.number.label("contract_number"))\
        .join(Contract, and_(Contract.id == subquery.c.id, subquery.c.state.in_(['Active','Process','Final','Retry'])))\
        .filter(subquery.c.rnk==1)
    return query

Contract_State_History类定义 #

class Contract_State_History(BaseModel):
    __tablename__ = '..'

    id = Column("contract_id", String, primary_key=True)
    _date = Column("period", DateTime, primary_key=True)
    state = Column("state_", String)

    @hybrid_property
    def date(self):
        return func.dateadd(text("month"), func.date_diff(text("month, 0"), self._date), 0)

首次执行错误 #

DatabaseException: Orig exception: Code: 43. DB::Exception: Second argument for function dateDiff must be Date or DateTime: While processing (0 + toIntervalMonth(dateDiff('month', 0, period))) <= '2022-12-07 16:17:49'. (ILLEGAL_TYPE_OF_ARGUMENT) (version 22.4.2.1 (official build))

修改混合属性后的错误 #

修改后的混合属性代码：

@hybrid_property
def date(self):
    return func.date_trunc('month', self._date) 

触发新错误：

DatabaseException: Orig exception: Code: 53. DB::Exception: Cannot convert string 2022-12-07 16:48:25 to type Date: while executing 'FUNCTION lessOrEquals(date_trunc('month', period) : 3, '2022-12-07 16:48:25' : 2) -> lessOrEquals(date_trunc('month', period), '2022-12-07 16:48:25') Nullable(UInt8) : 4'. (TYPE_MISMATCH) (version 22.4.2.1 (official build))

可行解决方案 #

方案1：统一类型转换，匹配Date类型 #

ClickHouse中date_trunc('month', DateTime)返回DateTime类型，需将两边统一转为Date类型避免不兼容：

• 修改混合属性，将截断结果转为Date：

  @hybrid_property
  def date(self):
      return func.toDate(func.date_trunc('month', self._date))
  

• 同步处理查询参数date为Date类型：

  from sqlalchemy import Date
  date = kwargs['date'] if '_date' in kwargs else datetime.today()
  date = func.toDate(date) if isinstance(date, datetime) else date
  

方案2：直接按月份范围过滤原始DateTime #

放弃混合属性，直接计算目标月份的起止时间，用原始_date字段过滤，绕开类型转换问题：

def get_active_contract(session: DBSession, **kwargs):
    target_date = kwargs['date'] if '_date' in kwargs else datetime.today()
    # 计算目标月份的起始和结束时间
    month_start = target_date.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    next_month = month_start.replace(month=month_start.month+1) if month_start.month <12 else month_start.replace(year=month_start.year+1, month=1)
    month_end = next_month - timedelta(microseconds=1)
    
    subquery = session.query(
        Contract_State_History.id.label("id"), 
        func.date_trunc('month', Contract_State_History._date).label("date"), 
        Contract_State_History.state.label("state"),
        func.rank().over(
            order_by=Contract_State_History._date.desc(),
            partition_by=Contract_State_History.id
        ).label("rnk")
    ).filter(Contract_State_History._date >= month_start, Contract_State_History._date <= month_end).subquery()
    
    # 后续query逻辑保持不变
    query = session.query(subquery.c.id.label("contract_id"), 
                        Contract.code.label("contract_code"), 
                        Contract.name.label("contract_name"), 
                        subquery.c.state.label("contract_state"),
                        Contract.number.label("contract_number"))\
        .join(Contract, and_(Contract.id == subquery.c.id, subquery.c.state.in_(['Active','Process','Final','Retry'])))\
        .filter(subquery.c.rnk==1)
    return query

方案3：使用ClickHouse原生函数toStartOfMonth #

ClickHouse原生toStartOfMonth函数返回月初DateTime，兼容性优于date_trunc：

• 修改混合属性：

  @hybrid_property
  def date(self):
      return func.toStartOfMonth(self._date)
  

• 调整过滤逻辑，直接匹配目标月份的月初：

  # 在get_active_contract中计算目标月份的月初
  target_month_start = func.toStartOfMonth(date)
  # 修改过滤条件为匹配目标月初
  subquery = session.query(...).filter(Contract_State_History.date == target_month_start).subquery()
  

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