CODEWITHSUNDEEP
sqlalchemyfastapialembic
jaspionU2
jaspionU2

Reputation: 31

create migrations with alembic

I am facing an issue when trying to generate migrations using Alembic and SQLAlchemy. My project is structured as follows:

  1. I have a folder called model where each entity (model) is defined in a separate file.

  2. I use a config folder containing a file called register.py, where I initialize an instance of registry() and store it in a variable named table_register.

  3. In my models, I use the @table_register.mapped_as_data_class decorator to map the classes to database tables.

The problem occurs when I run the command:

The revision file is generated, but both the upgrade and downgrade functions are empty. It seems that Alembic is not detecting the tables mapped in my project.

register.py file:

from sqlalchemy.orm import registry

table_register = registry()

metadata = table_register.metadata

Example model:

from enum import Enum

from datetime import datetime

import sqlalchemy

from sqlalchemy import (String, func, ForeignKey)
from sqlalchemy.orm import (Mapped, mapped_column)

from configs.register import table_register

@table_register.mapped_as_dataclass
class Aluno():
    __tablename__ = 'aluno'
    
    id: Mapped[int] = mapped_column(primary_key=True, init=False, name='id_aluno')
    nome: Mapped[str] = mapped_column(String(100), nullable=False, name='nome_aluno')
    serie: Mapped[int] = mapped_column(ForeignKey('serie.id_serie'), nullable=False, name='serie_aluno')
    turma: Mapped[Enum] = mapped_column(sqlalchemy.Enum('A', 'B', name='turma_enum'), nullable=False, name='turma_aluno')
    created_at: Mapped[datetime] = mapped_column(name='created_at', server_default=func.now(), init=False)

Alembic Config:

config = context.config
config.set_main_option('sqlalchemy.url', Config().DB_URI)

target_metadata = table_register.metadata

def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, 
            target_metadata=target_metadata,
            compare_type=True
        )

        with context.begin_transaction():
            context.run_migrations()

Even though everything seems to be configured correctly, Alembic does not detect the tables and generates empty migrations. I am unsure if I am missing something in the Alembic configuration or if there is a specific requirement for using SQLAlchemy's registry(). What could be causing this issue? Is there any additional configuration needed to make Alembic recognize tables defined using registry.mapped_as_data_class?

Upvotes: 0

Views: 89

Answers (1)

Mark
Mark

Reputation: 302

I had this same issue a couple years ago when first using alembic. Now I do not remember the exact solution but this is my config.

alembic/env.py

from os import getenv
from dotenv import load_dotenv
load_dotenv()

import sys
import os.path
# append parent dir to path to expose imports
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))

# db.py
from db import DB_BASE

from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

config = context.config

uri = getenv('DATABASE_URI')
if uri.startswith('postgres://'):
    uri = uri.replace('postgres://', 'postgresql://', 1)
config.set_main_option("sqlalchemy.url", uri)

fileConfig(config.config_file_name)

target_metadata = DB_BASE.metadata

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        render_as_batch=True,
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

db.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

uri = "YOUR_URI"

DB_ENGINE = create_engine(uri)
DB_SES_LOCAL = sessionmaker(bind=DB_ENGINE, autoflush=False)
DB_BASE = declarative_base()

from models import *

DB_BASE.metadata.create_all(DB_ENGINE)

Upvotes: 0

Related Questions