Cascade Delete in SqlAlchemy One-to-Many relationship made without relationships()

In my FastAPI app, I use sqlalchemy. The main goal of the app is that each user is stored in a sqlalchemy table users and can create different contents (25+ different contents).

Each content has its own table, and of course the goal is to link them to the users with a one to many relationship (one user can have multiple contents of each type).

Then I've seen that the proper way to do so is to create something like this :

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    content1 = relationship("Content1", backref='user')

class Content1(Base):
    __tablename__ = "content1"
    id = Column(Integer(), primary_key = True)
    user_id = Column(Integer(), ForeignKey('users.id'))

However, as I have many different contents and that this number could grow in the future, if I do like so then the User table will become really heavy and I'll need to update it everytime.

To fix this, my system is implemented like this :

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)

class Content1(Base):
    __tablename__ = "content1"
    id = Column(Integer(), primary_key = True)
    user_id = Column(Integer(), ForeignKey('users.id'))

As you can see I don't have any relationship in user. My first question is : is it even ok to do so or not recommended ? Indeed, I did this naturally as I'll never try to get a content directly from the user class but always by querying the db with the condition user_id == User.id

Then if working like this is ok I have a second question : how can I implement a cascade delete ? Because i want that when I delete a User, all the contents (of all my different content classes) linked to this user with user_id are deleted.

Thanks in advance !

Regards,

Cesar

If your database supports cascades natively then you can do this:

class Content1(Base):
    # ...
    user_id = Column(Integer(), ForeignKey('users.id', ondelete="CASCADE")))

NOTE That you have to have this set when you create the table otherwise you have to add it with ALTER TABLE either manually or with something like alembic.

Using type() to generate contents

If you know the content types ahead of time you can dynamically create the classes using type() like below. You can also use backref from the other direction so User is not changed directly. Using backref() can configure the relationship going in the other direction, here it is needed to prevent NULLing out the user_id.

import os

from sqlalchemy import (
    Column,
    Integer,
    create_engine,
    ForeignKey,
)
from sqlalchemy.sql import (
    select,
)
from sqlalchemy.orm import (
    backref,
    Session,
    relationship,
    declarative_base
)


def get_engine(env):
    return create_engine(
        f"postgresql+psycopg2://{env['DB_USER']}:{env['DB_PASSWORD']}@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}",
        echo=True,
    )


def get_plural_content_name(content_name):
    # Content1 -> content1s
    return f"{content_name.lower()}s"

Base = declarative_base()

content_names = ("Content1", "Content2", "Content3")

content_classes = {}

for name in content_names:
    plural_name = get_plural_content_name(name)
    content_props = {
        '__tablename__': plural_name,
        'id': Column(Integer, primary_key=True),
        'user_id': Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
        'user': relationship('User', backref=backref(plural_name, passive_deletes='all')),
    }

    cls = globals()[name] = content_classes[name] = type(name, (Base,), content_props)


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)


def main():
    engine = get_engine(os.environ)

    with engine.begin() as conn:
        Base.metadata.create_all(conn)
        populate(conn)
        query(conn)


def query(conn):
    with Session(conn) as session:
        u1 = session.scalars(select(User).limit(1)).first()
        for name in content_names:
            # Access the property user.content1s, user.content2s, etc.
            contents = getattr(u1, get_plural_content_name(name))
            assert len(contents) > 0
            assert contents[0].user == u1
        session.delete(u1)
        session.commit()
    with Session(conn) as session:
        assert not session.scalars(select(Content1).limit(1)).first()




def populate(conn):
    with Session(conn) as session:
        u1 = User()
        session.add(u1)
        for name in content_names:
            c = content_classes[name](user=u1)
            session.add(c)
        session.commit()


if __name__ == "__main__":
    main()