如何在 sqlalchemy 中使用 SQL Server 的 CAST(datetime as date) 函数？答案

【问题标题】：How to use function - CAST(datetime as date) of SQL Server in sqlalchemy?如何在 sqlalchemy 中使用 SQL Server 的 CAST(datetime as date) 函数？
【发布时间】：2015-12-28 05:05:11
【问题描述】：

我有一个包含列的表 - id(int)、date(DateTime)、value(float)。我想按日期输出具有id=2. 的所有行的每日平均分组

我为此编写了以下 sql 查询并且查询工作正常：

select cast(date as date), avg(value) from table where id=2 group by cast(date as date)

对于同样的问题，如果我想在 sqlalchemy 中执行，我使用以下表达式：

>> from sqlalchemy import sql, types
>> x = sql.cast(Table.date, types.Date)

但是当我尝试执行 x 时，出现以下错误：

>> x.execute()    
ArgumentError: Not an executable clause: CAST(Table.date AS DATETIME)

这意味着 SQL Server 正在考虑 types.Date as DateTime。请忽略 ArgumentError。我要关注的是内部sql查询使用CAST(Table.date AS DATETIME)而不是CAST(Table.date AS DATE) 有什么办法可以让我在 sqlalchemy 中将 DateTime 转换为 Date？

【问题讨论】：

您应该使用sqlalchemy.func.cast 而不是sqlalchemy.sql.cast。
我试过用它。我收到了这个错误：- ProgrammingError: (ProgrammingError) ('Invalid parameter type. param-index=0 param-type=Date') 'SELECT cast(Table.date, ?) AS cast_1, avg(Table.value) AS avg_1 FROM Table AS Table_1 WHERE Table_1.id = ? GROUP BY cast(Table_1.date, ?)' [Date(), 2, Date()]

标签： python sql-server sqlalchemy

【解决方案1】：

你的代码是错误的：

您不能执行clause。您需要有一个语句或一个查询（见下文）。这就是您最初收到错误的原因。
确实，如果您转换为sa.Date，mssql 引擎将转换为DATETIME。我不完全确定这是为什么，但我怀疑这是因为旧版本的 MSSQL 没有单独的 DATE-only 数据类型。要解决这个问题，只需将其转换为 mssql 方言中的 DATE 数据类型。

下面的代码应该可以工作。我现在没有mssql，但我确实展示了mssql 的SQL 语句的编译：

from sqlalchemy import create_engine, Column, Integer, Float, Date, DateTime, func
from sqlalchemy.dialects.mssql import DATE
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# @TODO: replace with your connection string
engine = create_engine('sqlite:///:memory:', echo=True)
Session = sessionmaker(bind=engine)
session = Session()

Base = declarative_base()


class MyTable(Base):
    __tablename__ = 'my_table'
    id = Column(Integer, primary_key=True)
    date = Column(DateTime())
    value = Column(Float())

Base.metadata.create_all(engine)

def get_mssql_sql(qry):
    from sqlalchemy.dialects import mssql
    dialect = mssql.dialect()
    return str(qry.compile(dialect=dialect))


# DATE = Date  # @note: this will not work in MSSQL (shows problem-2)
q = (session
     .query(func.cast(MyTable.date, DATE), func.avg(MyTable.value))
     .filter(MyTable.id == 2)
     .group_by(func.cast(MyTable.date, DATE))
     )
print(get_mssql_sql(q.statement))

# in your code, just run below to execute:
rows = q.all()

【讨论】：