SQLAlchemy - relationship proxy example

sqlalchemy_relationship_proxy.py

"""
Short example of how to set realtionship in many-to-many with extra column(s) in association table.
creator class is used on the relationship(), but it's also possible to create MUserPermitAssoc
explicitly and pass it to MUser.permits.

"""

from __future__ import annotations

import sqlalchemy as sa
import sqlalchemy.orm as so
from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

dsn = "some_dsn"

aengine = create_async_engine(dsn)
asession = async_sessionmaker(aengine)


class Base(DeclarativeBase):
    pass


class MUserPermitAssoc(Base):
    __tablename__ = "assoc_user_permit"

    user_id: so.Mapped[sa.UUID] = so.mapped_column(
        sa.ForeignKey("users.record_id", ondelete="CASCADE"),
        primary_key=True,
        nullable=False,
    )
    permit_id: so.Mapped[sa.UUID] = so.mapped_column(
        sa.ForeignKey("permits.record_id", ondelete="CASCADE"),
        primary_key=True,
        nullable=False,
    )

    user: so.Mapped["MUser"] = so.relationship(
        back_populates="permit_association", lazy="joined"
    )
    permit: so.Mapped["MPermit"] = so.relationship(
        back_populates="user_association", lazy="joined"
    )

    # Extra column
    assigned_by: so.Mapped[str | None] = so.mapped_column(default=None)


class MUser(Base):
    __tablename__ = "users"

    disabled: so.Mapped[bool] = so.mapped_column(sa.Boolean, default=False)
    first_name: so.Mapped[str] = so.mapped_column(sa.String(255))
    last_name: so.Mapped[str] = so.mapped_column(sa.String(255))
    middle_name: so.Mapped[str | None] = so.mapped_column(sa.String(255))
    username: so.Mapped[str] = so.mapped_column(sa.String(255), unique=True, index=True)

    permit_association: so.Mapped[list["MUserPermitAssoc"]] = so.relationship(
        back_populates="user"
    )
    permits: AssociationProxy[list[MPermit]] = association_proxy(
        "permit_association",
        "permit",
        creator=lambda permit_obj: MUserPermitAssoc(permit=permit_obj),
    )


class MPermit(Base):
    __tablename__ = "permits"

    value: so.Mapped[str] = so.mapped_column(unique=True, index=True)

    user_association: so.Mapped[list["MUserPermitAssoc"]] = so.relationship(
        back_populates="permit"
    )
    users: AssociationProxy[list["MUser"]] = association_proxy(
        "user_association", "user"
    )


# Create new Permit from User side using creator class on realtionship()
new_user = MUser("attr here")
new_permit = MPermit(value="abc")
new_user.permits.extend(new_permit)