Alembic migrations

database.py

# src/ex_machina/database.py

import click
import os

from alembic import command
from alembic.config import Config as AlembicConfig
from alembic.util import AutogenerateDiffsDetected
from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

from ex_machina.config import Config


engine = create_engine(Config.CONN_STRING)
autoengine = engine.execution_options(isolation_level="AUTOCOMMIT")
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))

meta = MetaData(
    naming_convention={
        "ix": "ix_%(column_0_label)s",
        "uq": "uq_%(table_name)s_%(column_0_name)s",
        "ck": "ck_%(table_name)s_%(column_0_name)s",
        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
        "pk": "pk_%(table_name)s"
    }
)

Base = declarative_base(metadata=meta)
Base.query = db_session.query_property()


def _get_alembic_config():
    directory = os.path.dirname(__file__)
    alembic_directory = os.path.join(directory, 'alembic')
    alembic_cfg = AlembicConfig()
    alembic_cfg.set_main_option('sqlalchemy.url', 
                                Config.CONN_STRING)
    alembic_cfg.set_main_option('script_location', alembic_directory)
    return alembic_cfg


@click.command()
def check():
    alembic_cfg = _get_alembic_config()
    command.check(alembic_cfg)


@click.command()
def migrate():
    # KEEP THIS IN SYNC WITH init_db() METHOD IN THIS FILE
    import ex_machina.admin.models
    import ex_machina.auth.models
    import ex_machina.consumables.models
    import ex_machina.departments.models
    import ex_machina.factory_settings.models
    import ex_machina.home.models
    import ex_machina.machines.models
    import ex_machina.products.models
    
    alembic_cfg = _get_alembic_config()
    
    try:
        command.check(alembic_cfg)
    except AutogenerateDiffsDetected:
        command.stamp(alembic_cfg, 'head')
        command.revision(alembic_cfg, autogenerate=True)


@click.command()
def upgrade():
    alembic_cfg = _get_alembic_config()
    command.upgrade(alembic_cfg, 'head')
    

@click.command()
def downgrade(hash_):
    alembic_cfg = _get_alembic_config()
    command.downgrade(alembic_cfg, hash_)


def init_db():
    # KEEP THIS IN SYNC WITH migrate() METHOD IN THIS FILE
    import ex_machina.admin.models
    import ex_machina.auth.models
    import ex_machina.consumables.models
    import ex_machina.departments.models
    import ex_machina.factory_settings.models
    import ex_machina.home.models
    import ex_machina.machines.models
    import ex_machina.products.models
    
    Base.metadata.create_all(bind=engine)

env.py

# src/alembic/env.py

from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
# target_metadata = None
from ex_machina import database
target_metadata = database.Base.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

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

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


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
        )

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


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

models.py

# src/ex_machina/consumables/models.py

from flask import current_app
from sqlalchemy import (
    BigInteger,
    Column,
    delete,
    Integer,
    ForeignKey,
    select, 
    String,
    Table,
    text
)
from sqlalchemy.orm import (Mapped, mapped_column)

from ex_machina.config import Config
from ex_machina.database import Base, autoengine, db_session
from ex_machina.util import natural_keys


class Consumable(Base):
    
    __tablename__ = 'consumables'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name = Column(String)
    code = Column(String, index=True)
    unit_measure = Column(ForeignKey('unit_measures.id'))
    unit_cost = Column(BigInteger) # Multiplied by CURRENCY_MULTIPLIER
    minimum_unit_purchase = Column(Integer)
    
    @staticmethod
    def create_new(
        
    ):
        return
    

product_consumables = Table(
    'product_consumables',
    Base.metadata,
    Column('parent', ForeignKey('products.id'), primary_key=True),
    Column('child', ForeignKey('consumables.id'), primary_key=True),
    Column('qty', BigInteger, nullable=False) # Multiplied by UNIT_MULTIPLIER
)