不使用 sqlalchemy_utils 创建 CompositeType 的 CompositeArray

Question

在我的 FastAPI、SQLAlchemy、Alembic 项目中，我使用的是 postgresql 引擎，现在我正尝试使用 postgresql+asyncpg 迁移到异步。这里的问题是我的数据库模式之一具有这种结构：

class MyTable(...):
    __tablename__ = 'my_table'

    name = Column(String(255), nullable=False, unique=True)
    tridimensional = Column(CompositeArray(
            CompositeType(
                'tridimensional_type', [
                    Column('x', Numeric(4, 0), nullable=False, default=0),
                    Column('y', Numeric(4, 0), nullable=False),
                    Column('z', Numeric(4, 0), nullable=False),
                ]
            ),
        ),
    )

因为这完全依赖于 sqlalchemy_utils.types.pg_composite（CompositeArray 和 CompositeType）并且不支持 register_composites for postgresql+asyncpg，我想知道（如果可能的话）如何：

创建我自己的 sqlalchemy.types.UserDefinedType 涉及此三维类型
将其添加到 alembic 迁移
在数组中使用它创建一个列

Answer 1

所以，我设法用使用 CompositeType 弄明白了：我创建了这两个 classes:

使用点运算符访问字典中的元素.

class Dotdict(dict):
    __getattr__ = dict.get
    __setattr__ = dict.__setitem__
    __delattr__ = dict.__delitem__

并从

继承的class进行处理

class AsyncCompositeType(CompositeType):
    # async sessions return asyncpg.Record not the type define in sqlalchemy_utils
    # these wrappers returns a dict when the composite type is loaded using an async session
    # see https://docs.sqlalchemy.org/en/14/core/custom_types.html#typedecorator-recipes
    def result_processor(self, dialect, coltype):
        def process(record):
            if isinstance(record, asyncpg.Record):  # handle async sqlalchemy session
                return Dotdict(record)
            return record
        return process

最后，您的 table 只会更改为：

class MyTable(...):
    __tablename__ = 'my_table'

    name = Column(String(255), nullable=False, unique=True)
    tridimensional = Column(CompositeArray(
            AsyncCompositeType(
                'tridimensional_type', [
                    Column('x', Numeric(4, 0), nullable=False, default=0),
                    Column('y', Numeric(4, 0), nullable=False),
                    Column('z', Numeric(4, 0), nullable=False),
                ]
            ),
        ),
    )

希望这对以后的人有所帮助。

不使用 sqlalchemy_utils 创建 CompositeType 的 CompositeArray

Create CompositeArray of CompositeType without using sqlalchemy_utils

python

sqlalchemy

alembic

asyncpg

fastapi