在 postgres 中使用 sqlalchemy 访问复合数据类型

Posted 2023-03-23

【中文标题】在 postgres 中使用 sqlalchemy 访问复合数据类型

【英文标题】：Access composite data type using sqlalchemy in postgres

【发布时间】：2017-09-25 20:19:42

【问题描述】：

我正在尝试使用 sqlalchemy 从 python 中的tiger.geocode 函数中提取复合列。 在纯 sql 形式中，它看起来像这样：

SELECT   
    g.rating  
    ,ST_X(g.geomout) As lon  
    ,ST_Y(g.geomout) As lat  
    ,(addy).address As stno  
    ,(addy).streetname As street  
    ,(addy).streettypeabbrev As styp  
    ,(addy).location As city  
    ,(addy).stateabbrev As st  
    ,(addy).zip  
FROM geocode(pagc_normalize_address('1 Capitol Square Columbus OH 43215')) As g  
;

这会产生以下输出：

#   rating  lon lat stno    street  styp    city    st  zip
1   17  -82.99782603089086  39.96172588526335   1   Capital St  Columbus    OH  43215

我面临的问题是如何在从 sqlalchemy（rating、lon、lat、stno、street、styp、city、st、zip）查询对象时引用复合列？

请，谢谢。

【问题讨论】：

你看过sqlalchemy_utils.types.pg_composite吗？

当我的复合类型是静态表上的列时，我可以看到它是如何工作的，但在我的例子中，它是由函数返回的复合类型。在我的设置中，我们没有为任何函数（仅表）定义 sqlalchemy 类。对这部分问题有什么想法吗？

【参考方案1】：

SQLAlchemy 不直接支持集合返回函数，但它的FunctionElements 被认为是FromClauses，这意味着您已经可以将它们视为表；我们只需要添加从函数中选择特定列的功能。幸运的是，这很简单（虽然不明显）：

from sqlalchemy.sql.base import ColumnCollection
from sqlalchemy.sql.expression import column
from sqlalchemy.sql.functions import FunctionElement

NormAddy = CompositeType(
    "norm_addy",
    [
        Column("address", Integer),
        Column("predirAbbrev", String),
        Column("streetName", String),
        Column("streetTypeAbbrev", String),
        Column("postdirAbbrev", String),
        Column("internal", String),
        Column("location", String),
        Column("stateAbbrev", String),
        Column("zip", String),
        Column("parsed", Boolean),
    ],
)

class geocode(GenericFunction):
    columns = ColumnCollection(
        Column("rating", Integer),
        column("geomout"),  # lowercase column because we don't have the `geometry` type
        Column("addy", NormAddy),
    )

GenericFunction 的子类化具有额外的好处，即全局注册 geocode 函数，以便 func.geocode 可以按预期工作。

g = func.geocode(func.pagc_normalize_address("1 Capitol Square Columbus OH 43215")).alias("g")
query = session.query(
    g.c.rating,
    func.ST_X(g.c.geomout).label("lon"),
    func.ST_Y(g.c.geomout).label("lat"),
    g.c.addy.address.label("stno"),
    g.c.addy.streetName.label("street"),
    g.c.addy.streetTypeAbbrev.label("styp"),
    g.c.addy.location.label("city"),
    g.c.addy.stateAbbrev.label("st"),
    g.c.addy.zip,
).select_from(g)

不幸的是，这并不完全奏效。似乎有一个错误使g.c.addy.address 语法在最新版本的 SQLAlchemy 上不起作用。我们可以快速修复它（尽管这应该在 sqlalchemy_utils 中修复）：

from sqlalchemy_utils.types.pg_composite import CompositeElement
import sqlalchemy_utils

class CompositeType(sqlalchemy_utils.CompositeType):
    class comparator_factory(_CompositeType.comparator_factory):
        def __getattr__(self, key):
            try:
                type_ = self.type.typemap[key]
            except KeyError:
                raise AttributeError(key)
            return CompositeElement(self.expr, key, type_)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.typemap = c.name: c.type for c in self.columns

现在可以了：

print(query.statement.compile(engine))
# SELECT g.rating, ST_X(g.geomout) AS lon, ST_Y(g.geomout) AS lat, (g.addy).address AS stno, (g.addy).streetName AS street, (g.addy).streetTypeAbbrev AS styp, (g.addy).location AS city, (g.addy).stateAbbrev AS st, (g.addy).zip AS zip_1 
# FROM geocode(pagc_normalize_address(%(pagc_normalize_address_1)s)) AS g