Sqlalchemy 关联查询

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship

# Create the database engine and session
engine = create_engine('sqlite:///join_demo.db')
Session = sessionmaker(bind=engine)
session = Session()

# Create the base model
Base = declarative_base()

# Define the table models
class Table1(Base):
    __tablename__ = 'table1'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    table2_rows = relationship("Table2")

class Table2(Base):
    __tablename__ = 'table2'
    id = Column(Integer, primary_key=True)
    table1_id = Column(Integer, ForeignKey('table1.id'))
    name = Column(String)

# Create the tables
Base.metadata.create_all(engine)

# Insert sample data
table1_data = [
    Table1(id=1, name='Table1 Row 1'),
    Table1(id=2, name='Table1 Row 2'),
    Table1(id=3, name='Table1 Row 3')
]
session.bulk_save_objects(table1_data)

table2_data = [
    Table2(id=1, table1_id=1, name='Table2 Row 1'),
    Table2(id=2, table1_id=3, name='Table2 Row 2')
]
session.bulk_save_objects(table2_data)

# 1. Inner Join
inner_join_query = session.query(Table1, Table2).join(Table2, Table1.id == Table2.table1_id)
inner_join_results = inner_join_query.all()
print("Inner Join Results:")
for result in inner_join_results:
    print(result[0].name, result[1].name)
print()

# 2. Left Join
left_join_query = session.query(Table1, Table2).join(Table2, Table1.id == Table2.table1_id, isouter=True)
left_join_results = left_join_query.all()
print("Left Join Results:")
for result in left_join_results:
    print(result[0].name, result[1].name if result[1] else "NULL")
print()

# 3. Right Join
right_join_query = session.query(Table1, Table2).join(Table2, Table1.id == Table2.table1_id, isouter=True).from_statement("RIGHT JOIN table1 ON table1.id = table2.table1_id")
right_join_results = right_join_query.all()
print("Right Join Results:")
for result in right_join_results:
    print(result[0].name if result[0] else "NULL", result[1].name)
print()

# 4. Full Join
full_join_query = session.query(Table1, Table2).join(Table2, Table1.id == Table2.table1_id, full=True)
full_join_results = full_join_query.all()
print("Full Join Results:")
for result in full_join_results:
    print(result[0].name if result[0] else "NULL", result[1].name if result[1] else "NULL")
print()

# 5. Cross Join
cross_join_query = session.query(Table1, Table2).cross_join(Table2)
cross_join_results = cross_join_query.all()
print("Cross Join Results:")
for result in cross_join_results:
    print(result[0].name, result[1].name)