Auto generate migrations alembic + SQLAlchemy imperative declaration

Question

I'm exploring a new DDD project using SQLAlchemy and Alembic for the first time. I'd like to use imperative mapping to isolate my domain objects. All the doc I can find about auto generating migrations with Alembic is using declarative mapping. Is it because I have to manually write all migrations if I want to use imperative mapping ?

Kalman

Answer 1

I had to import the metada of the Table I manually defined

Kalman

Just wanted to add to this that you need to import those manually defined metadatas and add them to the target_metadata = [imported_md1, imported_md2] for Alembic to see those. – Bonanza

Answer 2

I came to this page because autogenerating migrations no longer worked with the upgrade to SQLAlchemy 1.4, as the metadata were no longer recognized and the automatically generated migration deleted every table (DROP table in upgrade, CREATE TABLE in downgrade).

I have first tried to import the tables metadata like this :

target_metadata = [orm.table_company.metadata, orm.table_user.metadata, orm.table_3.metadata, orm.table_4.metadata] 

It resulted in the following error code:

alembic/autogenerate/api.py", line 462, in table_key_to_table
ValueError: Duplicate table keys across multiple MetaData objects: "tb_company", "tb_user", "tb_3", "tb_4"

I have found that rather than importing one metadata object per table, you can access it in a single pass with target_metadata = orm.mapper_registry.metadata :

SQL Alchemy 1.4

adapters/orm.py

from myapp import domain
from sqlalchemy.orm import registry
from sqlalchemy.schema import MetaData


metadata = MetaData()
mapper_registry = registry(metadata=metadata)

# define your tables here
table_user = Table(
    "tb_user",
    mapper_registry.metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column(
        "pk_user",
        UUID(as_uuid=True),
        primary_key=True,
        server_default=text("uuid_generate_v4()"),
        default=uuid.uuid4,
        unique=True,
        nullable=False,
    ),
    Column(
        "fk_company",
        UUID(as_uuid=True),
        ForeignKey("tb_company.pk_company"),
    ),
    Column("first_name", String(255)),
    Column("last_name", String(255)),
)


# map your domain objects to the tables
def start_mappers():
    mapper_registry.map_imperatively(
        domain.model.User,
        table_user,
        properties={
            "company": relationship(
                domain.Company,
                backref="users"
            )
         },
    )

alembic/env.py

from myapp.adapters import orm

# (...)
target_metadata = orm.mapper_registry.metadata

SQL Alchemy 1.3

Using classical / imperative mapping, Alembic could generate migrations from SQL Alchemy 1.3 with the following syntax:

adapters/orm.py

from myapp import domain
# (...)
from sqlalchemy import Table, Column
from sqlalchemy.orm import mapper, relationship
from sqlalchemy.schema import MetaData

metadata = MetaData()

# define your tables here
table_user = Table(
    "tb_user",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
     # (...)
)

# map your domain objects to the tables
def start_mappers():
 user_mapper = mapper(
        domain.User,
        tb_user,
        properties={
            "company": relationship(
                domain.Company,
                backref="users"
             ),
        },
    )

alembic/env.py

from myapp.adapters import orm

# (...)

target_metadata = orm.metadata

Rhinestone