Using association objects for many-to-many with extra columns

When the link between two entities has data of its own — a quantity, a role, a joined-at date — map the link table as its own class with two ForeignKey columns and a relationship() to each side, instead of relationship(secondary=...), and add an association_proxy where the far side is all the caller wants. This guide belongs to modeling relationships, cascades and association objects.

Quick Answer

A secondary table can only ever hold the two foreign keys. The moment the link needs a column, it has to become a class.

Before — a secondary table with nowhere to put the quantity:

from sqlalchemy import Column, ForeignKey, Table
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


order_products = Table(
    "order_products",
    Base.metadata,
    Column("order_id", ForeignKey("orders.id"), primary_key=True),
    Column("product_id", ForeignKey("products.id"), primary_key=True),
)


class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    products: Mapped[list["Product"]] = relationship(secondary=order_products)
    # Where does "three of this product, at the price on the day" go?

After — the link mapped as a class, with a proxy for convenience:

from sqlalchemy import ForeignKey, UniqueConstraint
from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class OrderItem(Base):
    __tablename__ = "order_items"
    __table_args__ = (UniqueConstraint("order_id", "product_id"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id", ondelete="CASCADE"))
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"))
    quantity: Mapped[int] = mapped_column(default=1)
    unit_price_cents: Mapped[int]

    order: Mapped["Order"] = relationship(back_populates="items")
    product: Mapped["Product"] = relationship(back_populates="order_items")


class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    items: Mapped[list[OrderItem]] = relationship(
        back_populates="order", cascade="all, delete-orphan", passive_deletes=True
    )
    products: AssociationProxy[list["Product"]] = association_proxy("items", "product")


class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    sku: Mapped[str]
    order_items: Mapped[list[OrderItem]] = relationship(back_populates="product")

order.items now gives the links with their data, and order.products still gives the products directly, so callers that never cared about quantity are unaffected.

Execution Context & Async Workflow Integration

An association object is not a special construct — it is an ordinary mapped class that happens to sit between two others, with a one-to-many relationship from each side into it. Everything you know about relationships applies unchanged, which is exactly why it scales better than secondary as requirements grow.

Writing goes through the association object. Because Order.items cascades save-update, appending an unsaved OrderItem is enough; SQLAlchemy inserts the order, then the item, filling order_id and product_id from the related objects, so no primary key has to be known in advance:

from shop.models import Order, OrderItem


async def add_line(session, order: Order, product, quantity: int) -> None:
    order.items.append(
        OrderItem(product=product, quantity=quantity, unit_price_cents=product.price_cents)
    )
    await session.commit()

Capturing unit_price_cents on the link is the point of the whole exercise: the price the customer paid belongs to the order, not to the product, and a secondary table has nowhere to keep it.

The association_proxy is a read-and-write convenience over the same data. Appending to order.products creates an OrderItem with only its product set, which works when every other column has a default and fails when one does not — so for models like this one, treat the proxy as read-only and write through order.items. A creator= callable makes the write path explicit when it is wanted: association_proxy("items", "product", creator=lambda p: OrderItem(product=p, unit_price_cents=p.price_cents)).

Under async, the thing to get right is loading depth. order.items is one relationship and item.product is another, so reading a product through an order is two levels, and both need loader options or the second raises MissingGreenlet. Chained loaders express that directly:

from sqlalchemy import select
from sqlalchemy.orm import selectinload

from shop.models import Order, OrderItem

stmt = (
    select(Order)
    .where(Order.id == order_id)
    .options(selectinload(Order.items).selectinload(OrderItem.product))
)
order = await session.scalar(stmt)

selectinload here costs two extra queries regardless of how many orders or items are involved, which is why it is the usual choice for collections; the trade-off against joinedload is covered in using selectinload vs joinedload for N+1 prevention. The association proxy follows whatever the underlying relationships loaded, so order.products is safe only once both levels are in memory.

Resolving Warnings, Errors & Common Mistakes

The "will copy column" warning is the most common one in this shape, and it is a genuine conflict rather than noise: two relationships both claim the right to write order_items.order_id, and which one wins depends on flush order. It gets a guide of its own — fixing "relationship will copy column" conflicts — because the same message appears in several other mappings.

The duplicate-key row is a design question disguised as an error. A unique constraint on (order_id, product_id) says a product appears at most once per order, so adding it again must mean "increase the quantity". Encode that in one place:

from sqlalchemy import select

from shop.models import OrderItem


async def add_or_merge_line(session, order, product, quantity: int) -> OrderItem:
    for item in order.items:                 # already loaded by the caller
        if item.product_id == product.id:
            item.quantity += quantity
            return item
    item = OrderItem(product=product, quantity=quantity,
                     unit_price_cents=product.price_cents)
    order.items.append(item)
    return item

Under concurrency this still races, and the resolution is the same as for any unique key: let the database decide, and handle the conflict — see handling IntegrityError on concurrent inserts.

Advanced: Querying Through the Link Without Loading Objects

Reports rarely want objects. When the question is "how many units of each product did we ship last month", the association table is just a table to join, and the ORM's job is to name the columns:

import datetime as dt

from sqlalchemy import func, select

from shop.models import Order, OrderItem, Product


def units_by_product(since: dt.date):
    return (
        select(
            Product.sku,
            func.sum(OrderItem.quantity).label("units"),
            func.sum(OrderItem.quantity * OrderItem.unit_price_cents).label("revenue_cents"),
        )
        .join(OrderItem, OrderItem.product_id == Product.id)
        .join(Order, Order.id == OrderItem.order_id)
        .where(Order.placed_on >= since, Order.status != "cancelled")
        .group_by(Product.sku)
        .order_by(func.sum(OrderItem.quantity).desc())
    )

Because unit_price_cents lives on the link, that revenue figure is what customers actually paid, which a join through a secondary table could not have produced at all.

Filtering parents by something about the link is the other common shape, and it wants any() rather than a join, so the parent rows are not duplicated:

from sqlalchemy import select

from shop.models import Order, OrderItem

# Orders containing more than five of any single product.
bulk_orders = select(Order).where(Order.items.any(OrderItem.quantity > 5))

# Orders that include a specific product at all.
with_product = select(Order).where(Order.items.any(OrderItem.product_id == product_id))

any() renders an EXISTS subquery, which returns each order once and can stop at the first matching item. A join would return one row per matching item and need distinct(), and combining it with eager loading of the same collection is what produces the cartesian-product warning described in fixing cartesian product warnings in SQLAlchemy joins.

When a filtered view of the collection is what you want — only the discounted items, say — contains_eager() with an explicit join lets the loaded collection be the filtered set rather than the whole one. That technique has its own guide in using contains_eager with filtered joins, and it comes with a warning worth repeating: the collection in memory then no longer represents every row in the database, so it must not be used to decide what to delete.

Migrating From a Secondary Table

Most association objects start life as a secondary table that outgrew itself. The migration is mechanical, and it can be done without downtime because the table itself does not change shape — only how the model sees it.

Step one: map the table as a class, without removing the old relationship. The class must describe the existing columns exactly, including the composite primary key if that is what the table has:

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship


class OrderProduct(Base):
    __tablename__ = "order_products"          # the existing secondary table

    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"), primary_key=True)
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), primary_key=True)

    order: Mapped["Order"] = relationship(back_populates="product_links")
    product: Mapped["Product"] = relationship(back_populates="order_links")

Step two: mark the old relationship viewonly=True. Two writable mappings over the same columns is exactly the conflict SQLAlchemy warns about, and viewonly resolves it while leaving every existing reader working:

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    products: Mapped[list["Product"]] = relationship(
        secondary="order_products", viewonly=True     # reads keep working
    )
    product_links: Mapped[list[OrderProduct]] = relationship(
        back_populates="order", cascade="all, delete-orphan"
    )

Step three: move the writers. Code that did order.products.append(product) becomes order.product_links.append(OrderProduct(product=product)). This is the only application change, and it can be done file by file while both mappings exist.

Step four: add the new columns. Only now does the schema change — a migration adding quantity and unit_price_cents, nullable at first, backfilled, then made NOT NULL if appropriate. The expand-and-contract sequence in adding a NOT NULL column without locking in Postgres applies unchanged.

Step five: replace the old relationship with an association_proxy, so order.products keeps working for readers without a second mapping of the same columns.

Doing it in this order means every deploy is compatible with the one before it, and the risky part — writers changing — happens while the schema is still exactly what it was.

Frequently Asked Questions

Related

• Modeling Relationships, Cascades and Association Objects — The parent guide: relationship shapes, loading and ownership.
• Configuring cascade delete and delete-orphan correctly — Making removal from the collection delete the link row.
• Fixing "relationship will copy column" conflict warnings — The warning two overlapping mappings produce.
• Typing relationships with Mappedlist in SQLAlchemy — Annotating the relationships an association object needs.