Create CompositeArray of CompositeType without using sqlalchemy_utils

Question

Within my FastAPI, SQLAlchemy, Alembic project, I was using postgresql engine, and now I'm trying to migrate to async with postgresql+asyncpg. The issue here is that one of my DB schemas has this structure:

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),
                ]
            ),
        ),
    )

Since this was relying entirely on sqlalchemy_utils.types.pg_composite (Both CompositeArray and CompositeType) and this does not have support to register_composites for postgresql+asyncpg, I was wondering (if possible) how to:

• Create my own sqlalchemy.types.UserDefinedType that involves this tridimensional type
• Add it to an alembic migration
• Create a column using it within an array

Answer 1

So, I managed to figure it out kinda using CompositeType: I created this two classes:

1. To access elements from a dict using the dot operator .

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

And the inherited class to process it from

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

Finally, your table would only be changed to:

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),
                ]
            ),
        ),
    )

Hopefully this helps someone in the future.