SQLAlchemy 和使用 numpy 数据类型进行过滤

Question

SQLAlchemy .filter() 函数似乎无法使用 numpy 数据类型。如果我在 filter 参数中使用 np.int32 ，则无法获得所需的结果。相反，我需要将我的 np.int32 转换为 int 以使查询按预期工作。

在下面的示例中，我从数据库中查询值，通过 numpy 进行一些选择，然后使用减少的选择再次查询（我知道这不是一个有意义的操作，只是为了演示问题）

import numpy as np

from sqlalchemy import Column, Integer, Float, ForeignKey
from sqlalchemy.ext.declarative import declarative_base

# ORM
Base = declarative_base()

class Example(Base):
    __tablename__ = 'Example'

    UID = Column(Integer, primary_key=True, autoincrement=True)
    Value = Column(Float)


from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///example.db', echo=False)
Session = sessionmaker(bind=engine)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

value = [1.1, 1.2, 2.1, 2.2]

session = Session()

for vi in value:
    example = Example(Value=vi)
    session.add(example)
    session.commit()


res = session.query(Example.UID, Example.Value).all()

uids = []
vals = []
for resi in res:
    uids.append(resi[0])
    vals.append(resi[1])

uids = np.array(uids) # creates a numpy array with elements of type np.int32
vals = np.array(vals)

idxsel = vals > 2
uids_sel = uids[idxsel]

for uidi in uids_sel:
    res = session.query(Example.Value).filter(Example.UID==uidi).all() # returns empty list
    print('without cast: {}'.format(res))
    res = session.query(Example.Value).filter(Example.UID==int(uidi)).all()
    print('with cast: {}'.format(res))

结果是

without cast: []
with cast: [(2.1,)] 
without cast: [] 
with cast: [(2.2,)]

我不确定，为什么 numpy 数据类型不起作用。我以前从未遇到过与 numpy 的任何兼容性问题。对我来说，这是一个相当严重的问题。当然你可以强制转换，但由于没有警告或错误，这很容易出现严重问题（意外丢失强制转换），因为如果你的查询不匹配，空列表当然是一个有效的响应......

由于 numpy 和 sqlalchemy 都相当普遍，如果它们兼容就太好了……

我真的很奇怪，我在谷歌的任何地方都没有提到这个问题。所以也许我错过了什么或做错了什么。任何帮助表示赞赏。

编辑： 问题在于 sqlite3 API 而不是 sqlalchemy。然而，在 sqlalchemy 的帮助下，我们可以使用 TypeDecorators 来解决这个问题：https://github.com/sqlalchemy/sqlalchemy/issues/3586

编辑： 这是一个关于如何在不使用 sqlalchemy 的情况下处理它的 stackoverflow inserting numpy integer types into sqlite with python3

Answer 1

我花了几个小时从不同的角度（where() 子句）遇到了相同的潜在问题。尽管最终编辑了上述问题，并提到了 TypeDecorator 类作为解决方案，但我想在这里实现该解决方案（使用 sqlalchemy 的核心 API 而不是 ORM）。希望页面上扩展的措辞改进了此问题的搜索结果。

类似的 MWE：

from pathlib import Path

import pandas as pd
from sqlalchemy import (
    create_engine,
    MetaData,
    Table,
    Column,
    Integer,
    String,
    select,
    insert,
)

eng = create_engine("sqlite:///"+str(Path().resolve() / 'temp.db'))
md = MetaData()
tbl = Table("users", md, Column("id", Integer), Column("name", String))
md.create_all(eng)
eng.execute(insert(tbl, (1,"Jack")))

stmt1 = select(tbl).where(tbl.c.id == 1)
print(eng.execute(stmt1).fetchall())

按预期返回：

[(1, 'Jack')]

但是：

target_id = pd.read_sql(stmt1, eng).loc[:,'id'].max()
stmt2 = select(tbl).where(tbl.c.id == target_id)
print(eng.execute(stmt2).fetchall())

返回：

[]

选择无法按预期运行，因为pandas.read_sql 将数据存储为 numpy 类型。

TypeDecorator解决方案

我们可以通过TypeDecorator 解决这个问题

import sqlalchemy.types as types

class IntegerLike(types.TypeDecorator):

    impl = types.Integer
    cache_ok = True

    def process_bind_param(self, value, dialect):
        return int(value)

如果我们删除上面刚刚创建的temp.db 并运行代码：

tbl = Table("users", md, Column("id", IntegerLike), Column("name", String))

诊断步骤？

我们可以通过

print(stmt1)

SELECT users.id, users.name 
FROM users 
WHERE users.id = :id_1

:id_1 是一个绑定参数，这些参数由底层 DBAPI 处理。 （我不知道如何真正强制 DBAPI 完全呈现语句，所以我欢迎任何提供方法的 cmets 或编辑）