Here are the 4 tables that I created using sqlalchemy.orm
from sqlalchemy.orm import Mapped, mapped_column, relationship, DeclarativeBase, MappedAsDataclass, Session, sessionmaker
from sqlalchemy import ForeignKey, create_engine
from sqlalchemy.dialects.postgresql import VARCHAR, INTEGER, NUMERIC
DATABASE_URL = 'postgresql://postgres:0000@localhost:5432/store'
engine = create_engine(url = DATABASE_URL)
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = Session()
class Base(DeclarativeBase, MappedAsDataclass):
pass
class User(Base):
__tablename__ = 'user'
id: Mapped[int] = mapped_column(INTEGER, primary_key=True, init=False)
username: Mapped[str] = mapped_column(VARCHAR, unique=True)
purchases: Mapped[list['Purchase']] = relationship(back_populates= 'user', init=False, cascade='all, delete')
class Manufacturer(Base):
__tablename__ = 'manufacturer'
id: Mapped[int] = mapped_column(INTEGER, primary_key=True, init=False)
name: Mapped[str] = mapped_column(VARCHAR)
products: Mapped[list['Product']] = relationship(back_populates='manufacturer', init=False)
class Product(Base):
__tablename__ = 'product'
id: Mapped[int] = mapped_column(INTEGER, primary_key=True, init=False)
name: Mapped[str] = mapped_column(VARCHAR, index=True)
manufacturer_id: Mapped[int] = mapped_column(ForeignKey('manufacturer.id'), init=False)
manufacturer: Mapped['Manufacturer'] = relationship(back_populates='products')
purchases: Mapped[list['Purchase']] = relationship(back_populates= 'product', init=False)
class Purchase(Base):
__tablename__ = 'purchase'
id: Mapped[int] = mapped_column(INTEGER, primary_key=True, init=False)
user_id: Mapped[int] = mapped_column(ForeignKey('user.id'), init=False)
product_id: Mapped[int] = mapped_column(ForeignKey('product.id'), init=False)
cost: Mapped[float] = mapped_column(NUMERIC(10, 2), index=True)
user: Mapped['User'] = relationship(back_populates='purchases')
product: Mapped['Product'] = relationship(back_populates='purchases')
This is the sequence in which I am running it
def create_user():
u = User(username = 'john')
session.add(u)
session.commit()
def create_manufacturer():
m = Manufacturer(name='apple')
session.add(m)
session.commit()
def create_product():
m = session.query(Manufacturer).where(Manufacturer.id == 1).first()
p = Product(name='iphone', manufacturer=m)
session.add(p)
session.commit()
def create_purchase():
p = session.query(Product).where(Product.id == 1).first()
u = session.query(User).where(User.id == 1).first()
pur = Purchase(cost = 999.99, product=p, user = u)
session.add(pur)
session.commit()
def show_user_purchases():
user = session.query(User).where(User.id == 1).first()
print(user.purchases)
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind = engine)
create_user()
create_manufacturer()
create_product()
create_purchase()
show_user_purchases()
Base on the output:
[Purchase(id=1, user_id=1, product_id=1, cost=Decimal('999.99'), user=User(id=1, username='john', purchases=[...]), product=Product(id=1, name='iphone', manufacturer_id=1, manufacturer=Manufacturer(id=1, name='apple', products=[...]), purchases=[...]))]
I suspect there is an infinite loop going on between user and purchases. Is there a way to fix it?
If you're using the dataclass integration, you'd want to set repr=False on
relationship attributes. Or your purchase will show user, which will show
purchases and so on, resulting in a loop (but I think the actual implementation
will show a ... if a loop is detected and omit the rest). Or you can use it
(repr=False) on the class and define your own __repr__ method instead of
getting it generated from the dataclass.
Updated code:
from sqlalchemy import ForeignKey, Integer, create_engine, select
from sqlalchemy.orm import (DeclarativeBase, Mapped, MappedAsDataclass,
mapped_column, relationship, sessionmaker)
DATABASE_URL = "postgresql+psycopg2://postgres:test_password@localhost:5432/test_db"
engine = create_engine(
DATABASE_URL,
)
Session = sessionmaker(
bind=engine,
)
class Base(MappedAsDataclass, DeclarativeBase):
pass
class User(Base):
__tablename__ = 'user'
id: Mapped[int] = mapped_column(Integer, primary_key=True, init=False)
purchases: Mapped[list['Purchase']] = relationship(
back_populates='user',
init=False,
cascade='all, delete',
repr=False,
)
class Manufacturer(Base):
__tablename__ = 'manufacturer'
id: Mapped[int] = mapped_column(Integer, primary_key=True, init=False)
products: Mapped[list['Product']] = relationship(
back_populates='manufacturer',
init=False,
repr=False,
)
class Product(Base):
__tablename__ = 'product'
id: Mapped[int] = mapped_column(Integer, primary_key=True, init=False)
manufacturer_id: Mapped[int] = mapped_column(
ForeignKey('manufacturer.id'),
init=False,
)
manufacturer: Mapped['Manufacturer'] = relationship(
back_populates='products',
repr=False,
)
purchases: Mapped[list['Purchase']] = relationship(
back_populates='product',
init=False,
repr=False,
)
class Purchase(Base):
__tablename__ = 'purchase'
id: Mapped[int] = mapped_column(Integer, primary_key=True, init=False)
user_id: Mapped[int] = mapped_column(ForeignKey('user.id'), init=False)
product_id: Mapped[int] = mapped_column(
ForeignKey('product.id'), init=False)
user: Mapped['User'] = relationship(
back_populates='purchases',
repr=False,
)
product: Mapped['Product'] = relationship(
back_populates='purchases',
repr=False,
)
if __name__ == "__main__":
Base.metadata.create_all(engine)
with Session() as session:
user = User()
session.add(user)
session.commit()
manufacturer = Manufacturer()
session.add(manufacturer)
session.commit()
product = Product(manufacturer=manufacturer)
session.add(product)
session.commit()
purchase = Purchase(user=user, product=product)
session.add(purchase)
session.commit()
with Session() as session:
user = session.scalar(select(User))
for purchase in user.purchases:
print(purchase)
This will show:
Purchase(id=1, user_id=1, product_id=1)