A example of custom json serializer:
# file: json.py
def custom_json_encoder(o):
if isinstance(o, pydantic.BaseModel):
res = o.dict()
return res['__root__'] if '__root__' in res else res
return pydantic_encoder(o)
def dumps(obj, **kwargs):
return json.dumps(obj, default=custom_json_encoder, **kwargs)
Bind it to the db engine:
create_async_engine(self.url, json_serializer=json.dumps, **engine_options)
-
Disable
sqlalchemy.url
setting inalembic.ini
. -
In
alembic/env.py
, something like:from myproj.conf import settings import myproj.models dbc = settings.db config = context.config config.set_main_option( 'sqlalchemy.url', f'postgresql+psycopg2://{dbc.user}:{dbc.password}@{dbc.host}:{dbc.port}/{dbc.database}' )