В моем проекте 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
для postgresql+asyncpg
, мне было интересно (если возможно), как:
sqlalchemy.types.UserDefinedType
, который включает в себя этот трехмерный типИтак, мне удалось выяснить это вроде с использованиемCompositeType
:
Я создал эти два класса:
.
class Dotdict(dict):
__getattr__ = dict.get
__setattr__ = dict.__setitem__
__delattr__ = dict.__delitem__
И унаследованный класс для его обработки
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
Наконец, ваша таблица будет изменена только на:
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),
]
),
),
)
Надеюсь, это поможет кому-то в будущем.