Optimizing Database Tables with Generics: Python and SQLAlchemy Approach

It's not always shortened to types, but indeed we can use this method to optimize our tables. In this article, we'll follow the structure of benefiting from generics in models.

Tech used in this article:

• Python

• FastAPI

• PostgreSQL

• SQLAlchemy

With understanding this methodology, we can easily apply this approach to any flexible frameworks or languages but before we dive in, it's better to know how it works first, with the help of visual tools.

1. ContentTypes : The place where it holds the identity of the models, and assign an Id to each model allowing us to later determine which table the data corresponds to.

2. GenericPictures : The primary table in the database that stores data using content_type_id and object_id. The object_id represents the ID of the record associated with the target table. In this scenario, we store the picture by its URL, but you can include any other fields instead.

3. PictureMixIn : A type of mixin when we added to the target table it inherit the generic feature.

4. Product : An example table for this scenario.

We collect the model names in this table

# where base comes from
# Base = declarative_base(metadata=metadata)

class ContentType(Base):
    __tablename__ = "content_types"

    model: Mapped[str]= mapped_column(String, nullable=False, unique=True)

We store the picture's URL, along with a content type ID (which tells us which table the data belongs to), and an object ID (the specific ID of the item in the target table).

class GenericPictures(Base):

    __tablename__ = 'generic_pictures'

    picture_url: Mapped[str] = mapped_column(String, nullable=False)
    content_type_id: Mapped[int] = mapped_column(Integer, ForeignKey('content_types.id'), nullable=False)
    object_id: Mapped[int] = mapped_column(Integer, nullable=False)

    content_type = relationship("ContentType")

When we add this mixin to a table, it brings along the generic features. It sets up a relation between genericPicture and targeted model. But the real magic happens with the setup function. This function triggers when we're migrating new data into the database (LIKE A NEW TABLE). It's called in env.py, which is a file for alembic migrations (we'll talk about that later). The setup function creates new instance data for the models in the table.

(JUST LIKE THE Django Content_Type table)

from sqlalchemy.orm import Session

class PictureMixin:

    @declared_attr
    def images(cls):
        model_name = cls.__tablename__
        content_type_id = (
            select(ContentType.id)
            .where(ContentType.model == model_name)
            .scalar_subquery()
        )
        return relationship(
            "GenericPictures",
            primaryjoin=lambda: and_(
                content_type_id == GenericPictures.content_type_id,
                cls.id == GenericPictures.object_id
            ),
            foreign_keys=[GenericPictures.content_type_id, GenericPictures.object_id],
            viewonly=True
        )


    # creating the initial data for each assigned model
    @classmethod
    def setup(cls, session: Session):
        if hasattr(cls, '__tablename__'):
            model_name = cls.__tablename__


            content_type = session.query(ContentType).filter_by(model = model_name).first()
            if not content_type:
                content_type = ContentType(model = model_name)
                session.add(content_type)
                session.commit()

            # to set the contenttype id for next usage
            cls.content_type_id = content_type.id
        else:
            pass

We inherit the PictureMixin to enable its features

class Product(Base, PictureMixin):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
    slug: Mapped[str] = mapped_column(String, unique=True, nullable=False)
    description: Mapped[str] = mapped_column(Text, nullable=True)
    price: Mapped[float] = mapped_column(Float, nullable=False)

This is where almost most magics begin Not all parts of the code generated by Alembic may be necessary, but I have added custom comments to the new sections for clarity. As you can see here on run_setup_on_migration function, we capture the context which i to us by the configuration in env.py. The 'on_version_apply' parameter helps us to run functions after each migration. It can be a list, so we can run multiple functions in order after a migration is applied to the database. In the run_setup_on_migrationfunction, we run the initial data creation after the initial tables have been created and migrated into the database. This function loops through all subclasses of PictureMixin, calling their setup methods to populate the intial data.

from logging.config import fileConfig
from time import sleep

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context
from common.config import settings
from common.config.settings import Settings
from common.models.pictureMixIn import PictureMixin
from common.db.session import DBSessionManager, metadata 
from common.db.session import Base

#? Custom models
from path_to_product import Product

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
#! Custom
settings = Settings()
# database connection
datbase_url = settings.FULL_DATABASE_PG_URL
config.set_main_option("sqlalchemy.url", database_url)

# 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
#? Custom
target_metadata = 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.

#? Custom
def run_setup_on_migration(ctx, **args):
    with context.begin_transaction():
        context.run_migrations()
    sleep(1)
    with DBSessionManager().session_sync() as session:
        for class_ in PictureMixin.__subclasses__():
            print(f"Running setup for {class_.__name__}")
            class_.setup(session)

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,
            #? Custom
            target_metadata=target_metadata ,
            on_version_apply=run_setup_on_migration
        )

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


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

Note: This code uses DBSessionManager().session_sync() to get a session. You'll need to implement your own method for creating and managing database sessions , and then pass that session to the setup method.

And after creating a sample data in the product table and passing the image for upload, this is the data captured in the generic_picture table.

📝 Conclusion:

This generic picture system we've set up can be pretty useful. It helps us avoid duplicating code across different tables. Instead of creating separate image fields for every table that needs pictures, which could get messy real quick - so we create one central spot where all the image info comes from. One major benefit that stands out especially as the app grows and data handling becomes more complex, is the setup and its flexibility. We used photo URLs in this example, but you can easily adapt it to store whatever information you might need.

Sources:

Alembic doc: https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.e

Github Repository: https://github.com/itz-Amethyst/fastapi-whistler-microservices