SqlAlchemy relations example

main.py

# sqlalchemy model with order and order_item tables
# sqlite3 database
# SQLAlchemy version 1.4
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, registry


mapper_registry = registry()
Base = mapper_registry.generate_base()

class Order(Base):
    __tablename__ = 'order'
    id = Column(Integer, primary_key=True)
    customer = Column(String, unique=True)
    items = relationship('OrderItem', back_populates='order')
    shipping = relationship('Shipping', uselist=False, back_populates='order')


class OrderItem(Base):
    __tablename__ = 'order_item'
    id = Column(Integer, primary_key=True)
    order_id = Column(Integer, ForeignKey('order.id'))
    order = relationship('Order', back_populates='items')
    item = Column(String)

# Shipping status enitity
class Shipping(Base):
    __tablename__ = 'shipping'
    id = Column(Integer, primary_key=True)
    order_id = Column(Integer, ForeignKey('order.id'))
    order = relationship('Order', back_populates='shipping')
    status = Column(String)


def create_session():
    engine = create_engine('sqlite:///order.db')
    # engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine)

    from sqlalchemy.orm import sessionmaker
    Session = sessionmaker(bind=engine)

    session = Session()
    return session

def add_and_query(session):
    # Add three orders with some items on each order
    order1 = Order(customer='Nancy')
    order1.items.extend([OrderItem(item='Apple'), OrderItem(item='Banana')])
    session.add(order1)

    order2 = Order(customer='John')
    order2.items.extend([OrderItem(item='Orange'), OrderItem(item='Pineapple')])
    session.add(order2)

    name = input('Enter the name of the customer: ')
    order3 = Order(customer=name)
    order3.items.extend([OrderItem(item='Strawberry'), OrderItem(item='Watermelon'), OrderItem(item='Kiwi')])
    session.add(order3)
    order3.items.append(OrderItem(item='Cloudberry'))

    session.commit()

    # Query all orders
    orders = session.query(Order).all()
    for order in orders:
        print(order.id, order.customer)
        for item in order.items:
            print('  ', item.item)
        print()


    # Delete John's order
    session.query(Order).filter(Order.customer == 'John').delete()
    session.commit()

    # Ship Nancy's order
    order = session.query(Order).filter(Order.customer == 'Nancy').one()
    shipping = Shipping(status='Shipped')
    order.shipping = shipping
    session.add(shipping)
    session.commit()

    # Check if Nancy have any order
    order_count = session.query(Order).filter(Order.customer == 'Nancy').count()
    print(f'Nancy has ordes {order_count}')

    # Query all orders with items and shipping status
    orders = session.query(Order).join(OrderItem).outerjoin(Shipping).all()
    for order in orders:
        print(order.id, order.customer)
        for item in order.items:
            print('  ', item.item)
        if order.shipping:
            print('  Shipping status:', order.shipping.status)
        print()


if __name__ == '__main__':
    session = create_session()
    add_and_query(session)