Demonstration of lookup tables with SQLAlchemy that automatically update with new entries

sqlalchemy_enum.py

import itertools
import uuid
from functools import lru_cache
from typing import Dict, List, Optional, Union

from sqlalchemy import (Column, DateTime, ForeignKey, Integer, String,
                        create_engine)
from sqlalchemy.dialects import mysql
from sqlalchemy.orm import (Session, declarative_base, relationship,
                            sessionmaker)
from sqlalchemy.schema import MetaData
from sqlalchemy.sql import func
from sqlalchemy.types import TypeDecorator
from tabulate import tabulate

conn_params: Dict[str, Union[str, int]] = {
    "host": "127.0.0.1",
    "user": "root",
    "password": "abc",
    "database": "foo",
    "port": 3306,
}


Metadata = MetaData(
    naming_convention={
        "fk": "%(table_name)s_%(column_0_name)s_%(referred_table_name)s_fk",
        "ix": "%(table_name)s_%(column_0_label)s_idx",
        "pk": "%(table_name)s_pk",
        "uq": "%(table_name)s_%(column_0_label)s_unq",
    }
)
Base = declarative_base(metadata=Metadata)


class BinaryUUID(TypeDecorator):
    impl = mysql.BINARY(16)

    def process_bind_param(self, value, dialect):
        try:
            return value.bytes
        except AttributeError:
            try:
                return uuid.UUID(bytes=value)
            except TypeError:
                return value

    def process_result_value(self, value, dialect):
        return uuid.UUID(bytes=value)


class TxnType(Base):
    __tablename__ = "TxnType"

    id = Column(mysql.TINYINT(unsigned=True), primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False, unique=True)
    txn_type = relationship("EnumAsLookup")


class TxnStatus(Base):
    __tablename__ = "TxnStatus"

    id = Column(mysql.TINYINT(unsigned=True), primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False, unique=True)
    txn_status = relationship("EnumAsLookup")


class EnumAsLookup(Base):
    __tablename__ = "EnumAsLookup"

    id = Column(Integer, primary_key=True, autoincrement=True)
    created_at = Column(DateTime, nullable=False, server_default=func.now())
    txn_id = Column(BinaryUUID, nullable=False, default=uuid.uuid4)
    txn_type_id = Column(
        mysql.TINYINT(unsigned=True), ForeignKey("TxnType.id"), nullable=False
    )
    txn_status_id = Column(
        mysql.TINYINT(unsigned=True), ForeignKey("TxnStatus.id"), nullable=False
    )


@lru_cache(maxsize=255)
def get_txn_status_id(session: Session, txn_status_name: str) -> Optional[int]:
    if (
        res := session.query(TxnStatus)
        .filter(TxnStatus.name == txn_status_name)
        .first()
    ) is not None:
        return res.id
    return None


@lru_cache(maxsize=255)
def get_txn_type_id(session: Session, txn_type_name: str) -> Optional[int]:
    if (
        res := session.query(TxnType).filter(TxnType.name == txn_type_name).first()
    ) is not None:
        return res.id
    return None


def set_txn_status_id(session: Session, txn_status_name: str) -> int:
    txn_status_entry = TxnStatus(name=txn_status_name)
    session.add(txn_status_entry)
    session.commit()
    return txn_status_entry.id


def set_txn_type_id(session: Session, txn_type_name: str) -> int:
    txn_type_entry = TxnType(name=txn_type_name)
    session.add(txn_type_entry)
    session.commit()
    return txn_type_entry.id


def insert_new_row_to_enum_table(
    session: Session, txn_type: str, txn_status: str
) -> int:
    if (_txn_type_id := get_txn_type_id(session, txn_type)) is not None:
        pass
    else:
        _txn_type_id = set_txn_type_id(session, txn_type)
    if (_txn_status_id := get_txn_status_id(session, txn_status)) is not None:
        pass
    else:
        _txn_status_id = set_txn_status_id(session, txn_status)

    enum_as_lookup_entry = EnumAsLookup(
        txn_type_id=_txn_type_id, txn_status_id=_txn_status_id
    )
    session.add(enum_as_lookup_entry)
    return enum_as_lookup_entry.id


def bulk_insert_to_enum_table(
    session: Session, txn_types: List, txn_statuses: list
) -> None:
    _txn_type_ids: List = []
    _txn_status_ids: List = []
    for txn_type in txn_types:
        if _txn_type_id := get_txn_type_id(session, txn_type.name) is not None:
            _txn_type_ids.append(_txn_type_id)
        else:
            _txn_type_ids.append(set_txn_type_id(session, txn_type.name))

    for txn_status in txn_statuses:
        if _txn_status_id := get_txn_status_id(session, txn_status.name) is not None:
            _txn_status_ids.append(_txn_status_id)
            pass
        else:
            _txn_status_ids.append(set_txn_status_id(session, txn_status.name))

    enum_as_lookup_entries = [
        EnumAsLookup(
            txn_type_id=txn_type_id,
            txn_status_id=txn_status_id,
        )
        for txn_type_id, txn_status_id in itertools.product(
            _txn_status_ids, _txn_status_ids
        )
    ]

    session.add_all(enum_as_lookup_entries)
    session.commit()


if __name__ == "__main__":
    engine = create_engine("mysql+pymysql://", connect_args=conn_params)
    TxnType.metadata.create_all(engine, checkfirst=True)
    TxnStatus.metadata.create_all(engine, checkfirst=True)
    EnumAsLookup.metadata.create_all(engine, checkfirst=True)
    MySession = sessionmaker(bind=engine)
    mysession = MySession()

    txn_statuses: List = [
        TxnStatus(name="SUCCEEDED"),
        TxnStatus(name="FAILED"),
        TxnStatus(name="REVERSED"),
    ]

    txn_types: List = [
        TxnType(name="OPEN"),
        TxnType(name="SOLD"),
        TxnType(name="CLOSED"),
        TxnType(name="PAYMENT"),
    ]

    bulk_insert_to_enum_table(
        session=mysession, txn_types=txn_types, txn_statuses=txn_statuses
    )

    result = [
        {col: getattr(txn, col) for col in EnumAsLookup.__table__.columns.keys()}
        for txn in mysession.query(EnumAsLookup).all()
    ]
    print(tabulate(result, headers="keys", tablefmt="psql"))

# output: 
"""
❯ python3 test_enums.py
+------+---------------------+--------------------------------------+---------------+-----------------+
|   id | created_at          | txn_id                               |   txn_type_id |   txn_status_id |
|------+---------------------+--------------------------------------+---------------+-----------------|
|    1 | 2024-12-20 13:39:18 | 305036d9-38be-4e8b-bde2-bfbc4b228954 |             1 |               1 |
|    2 | 2024-12-20 13:39:18 | 93c2da11-3279-4064-87f8-d6b00254bf01 |             1 |               2 |
|    3 | 2024-12-20 13:39:18 | 7251bee6-809f-4afa-b4ce-938c89b9ab52 |             1 |               3 |
|    4 | 2024-12-20 13:39:18 | 07f8b38c-e4d0-4081-83c1-eda2d3e9cb8e |             2 |               1 |
|    5 | 2024-12-20 13:39:18 | 1817f30d-beb5-4809-8013-8f28f4ef222a |             2 |               2 |
|    6 | 2024-12-20 13:39:18 | a87e2681-56de-4522-b057-13e39d0e469f |             2 |               3 |
|    7 | 2024-12-20 13:39:18 | ba30343d-b075-4235-aa83-ff268ed4fd8f |             3 |               1 |
|    8 | 2024-12-20 13:39:18 | a7520c1d-fcfe-4e43-ba05-2efcb9331ff5 |             3 |               2 |
|    9 | 2024-12-20 13:39:18 | ec6a244a-e172-4891-93e8-529a40b71336 |             3 |               3 |
+------+---------------------+--------------------------------------+---------------+-----------------+
"""