如何一致地从 SQLAlchemy 获取非标准类型的 python_type ？

Question

我正在尝试创建一个函数，通过 SQLAlchemy 表达式语言将 Python 字典作为一行写入 Postgres 表。我想检查该字典中传递的值是否与数据库表的相应列类型匹配。

到目前为止我有类似的东西

import sqlalchemy
from sqlalchemy import MetaData

def create(table, new_row):
    engine = sqlalchemy.create_engine(~my db address~) 
    config_metadata = MetaData()
    config_metadata.reflect(engine)
    target_table = config_metadata.tables[table]
    target_column_types = {col.name:col.type.python_type for col in target_table.c}
    # ... Code to check that target_column_types match types of new_row

根据文档 python_type给出了预期返回的相应类型，例如intInteger 和dictJSONB。但是，如果我在 INT4RANGE Postgres 列上调用它，我会得到一个NotImplementedError(). 根据经验，我可以使用 SQLAlchemy 查询同一范围列并返回结果，其类型为psycopg2._range.NumericRange

那么，为什么不.python_type直接调用该列给我呢？而且，在未实现的psycopg2._range.NumericRange情况下，我可以做些什么来检索给定的预期返回类型吗？python_type

Answer 1

我通过创建自定义解决了这个问题IntRangeType，并让我的模型将其用于列定义。

import intervals
from sqlalchemy.dialects.postgresql import INT4RANGE
from sqlalchemy_utils import IntRangeType as _IntRangeType

class IntRangeType(_IntRangeType):
    class INT4RANGE(INT4RANGE):
        python_type = type(intervals.IntInterval)

impl = INT4RANGE
comparator_factory = _IntRangeType.comparator_factory
cache_ok = True

def __init__(self, *args, **kwargs):
    super(IntRangeType, self).__init__(*args, **kwargs)
    self.interval_class = intervals.IntInterval