SQLAlchemyを使って外部制約キーを複数持ったリレーションシップを設定する
リレーションシップの基本
構築したいDBのER図
CREATE TABLE persons (
id VARCHAR(4) NOT NULL,
name VARCHAR(30) NOT NULL,
PRIMARY KEY (id)
)
CREATE TABLE tickets (
person_id VARCHAR(4) NOT NULL,
ticket_id VARCHAR(8) NOT NULL,
room_name VARCHAR(30) NOT NULL,
PRIMARY KEY (person_id, ticket_id),
FOREIGN KEY(person_id) REFERENCES persons (id)
)
CREATE TABLE bookings (
person_id VARCHAR(4) NOT NULL,
ticket_id VARCHAR(8) NOT NULL,
booking_id VARCHAR(12) NOT NULL,
comment VARCHAR(30),
PRIMARY KEY (person_id, ticket_id, booking_id),
FOREIGN KEY(person_id, ticket_id) REFERENCES tickets (person_id, ticket_id)
)
環境
- Python 3.8.10
- SQLAlchemy == 1.4.27
基本的なリレーションシップのパターン
ER図
- Personクラスからrelation_shipでTicketクラスにリレーションを貼る
- Person側のrelation_shipにback_refがついているのでTicket側にはrelation_shinpの設定をしなくてもリレーションが貼られる
- Personのtickets = relationship("Ticket", backref="persons")の様にする
pythonのソース(全文)
from sqlalchemy import VARCHAR, Column, ForeignKey, MetaData, UniqueConstraint, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, scoped_session, sessionmaker
url = "sqlite:///"
engine = create_engine(url, echo=False)
Session = scoped_session(sessionmaker(bind=engine, autocommit=False, autoflush=False))
Base = declarative_base()
class Person(Base):
__tablename__ = "persons"
id = Column(VARCHAR(4), autoincrement=False, primary_key=True)
name = Column(VARCHAR(30), nullable=False)
tickets = relationship("Ticket", backref="persons")
def dict(self):
tickets = [ticket.dict() for ticket in self.tickets]
res = {"id": self.id, "name": self.name, "tickets": tickets}
return res
class Ticket(Base):
__tablename__ = "tickets"
person_id = Column(VARCHAR(4), ForeignKey("persons.id"), primary_key=True, autoincrement=False)
ticket_id = Column(VARCHAR(8), primary_key=True, autoincrement=False)
room_name = Column(VARCHAR(30), nullable=False)
def dict(self):
res = {
"person_id": self.person_id,
"ticket_id": self.ticket_id,
"room_name": self.room_name,
}
return res
metadata = MetaData()
Base.metadata.create_all(bind=engine, checkfirst=True)
def insert_data():
with Session() as session:
person = Person(id=1, name="hoge")
tickets = [
Ticket(person_id="1000", ticket_id="10001000", room_name="hoge"),
Ticket(person_id="1000", ticket_id="10001001", room_name="hage"),
]
person.tickets.extend(tickets)
session.add(person)
session.commit()
def query():
import json
with Session() as session:
session = Session()
res = session.query(Person).one()
print(json.dumps(res.dict(), indent=2))
res = session.query(Ticket).filter(Ticket.ticket_id == "10001000").one()
print(json.dumps(res.dict(), indent=2))
print(json.dumps(res.persons.dict(), indent=2))
if __name__ == "__main__":
insert_data()
query()
実行結果
# print(json.dumps(res.dict(), indent=2))
{
"id": "1",
"name": "hoge",
"tickets": [
{
"person_id": "1",
"ticket_id": "10001000",
"room_name": "hoge"
},
{
"person_id": "1",
"ticket_id": "10001001",
"room_name": "hage"
}
]
}
# print(json.dumps(res.dict(), indent=2))
{
"person_id": "1",
"ticket_id": "10001000",
"room_name": "hoge"
}
# リレーションを参照して親(Person)側から再度データを取得している
# print(json.dumps(res.persons.dict(), indent=2))
{
"id": "1",
"name": "hoge",
"tickets": [
{
"person_id": "1",
"ticket_id": "10001000",
"room_name": "hoge"
},
{
"person_id": "1",
"ticket_id": "10001001",
"room_name": "hage"
}
]
}
拡張したパターン
ER図(最初に貼ったのと同じ)
- ticketテーブルのときは外部制約に利用するperson_idにはForeignkeyをつけたけれどもbookinsの様に外部制約キーが2個ある場合はテーブルのColumnの方にはForeignKeyをつけずに table_args = (ForeignKeyConstraint(["person_id", "ticket_id"], ["tickets.person_id", "tickets.ticket_id"]),)の様にする
bookingとticketの双方向のリレーションを構築する
tikect → booking
bookings = relationship(
"Booking", foreign_keys="[Booking.person_id, Booking.ticket_id]", back_populates="ticket_back_populates"
)
booking → ticket
__table_args__ = (ForeignKeyConstraint(["person_id", "ticket_id"], ["tickets.person_id", "tickets.ticket_id"]),)
ticket_back_populates = relationship(
"Ticket", foreign_keys="[Booking.person_id, Booking.ticket_id]", back_populates="bookings"
)
pythonのソース(全文)
from sqlalchemy import VARCHAR, Column, ForeignKey, ForeignKeyConstraint, MetaData, UniqueConstraint, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, scoped_session, sessionmaker
url = "sqlite:///"
engine = create_engine(url, echo=False)
Session = scoped_session(sessionmaker(bind=engine, autocommit=False, autoflush=False))
Base = declarative_base()
class Person(Base):
__tablename__ = "persons"
id = Column(VARCHAR(4), autoincrement=False, primary_key=True)
name = Column(VARCHAR(30), nullable=False)
tickets = relationship("Ticket", backref="persons")
def dict(self):
tickets = [ticket.dict() for ticket in self.tickets]
res = {"id": self.id, "name": self.name, "tickets": tickets}
return res
class Ticket(Base):
__tablename__ = "tickets"
person_id = Column(VARCHAR(4), ForeignKey("persons.id"), primary_key=True, autoincrement=False)
ticket_id = Column(VARCHAR(8), primary_key=True, autoincrement=False)
room_name = Column(VARCHAR(30), nullable=False)
bookings = relationship(
"Booking", foreign_keys="[Booking.person_id, Booking.ticket_id]", back_populates="ticket_back_populates"
)
def dict(self):
bookings = [booking.dict() for booking in self.bookings]
res = {
"person_id": self.person_id,
"ticket_id": self.ticket_id,
"room_name": self.room_name,
"bookings": bookings,
}
return res
class Booking(Base):
__tablename__ = "bookings"
__table_args__ = (ForeignKeyConstraint(["person_id", "ticket_id"], ["tickets.person_id", "tickets.ticket_id"]),)
person_id = Column(VARCHAR(4), primary_key=True, autoincrement=False)
ticket_id = Column(VARCHAR(8), primary_key=True, autoincrement=False)
booking_id = Column(VARCHAR(12), primary_key=True, autoincrement=False)
comment = Column(VARCHAR(30), nullable=True)
ticket_back_populates = relationship(
"Ticket", foreign_keys="[Booking.person_id, Booking.ticket_id]", back_populates="bookings"
)
def dict(self):
res = {
"person_id": self.person_id,
"ticket_id": self.ticket_id,
"booking_id": self.booking_id,
"comment": self.comment,
}
return res
metadata = MetaData()
metadata.reflect(engine)
Base.metadata.create_all(bind=engine, checkfirst=True)
def insert_data():
with Session() as session:
person = Person(id=1, name="hoge")
tickets = [
Ticket(person_id="1000", ticket_id="10001000", room_name="hoge"),
Ticket(person_id="1000", ticket_id="10001001", room_name="hage"),
]
booking1 = Booking(person_id="1000", ticket_id="10001000", booking_id="100010001000", comment="first ticket")
tickets[0].bookings.append(booking1)
booking2 = [
Booking(person_id="1000", ticket_id="10001001", booking_id="100010011000", comment="first ticket"),
Booking(person_id="1000", ticket_id="10001001", booking_id="100010011001", comment="second ticket"),
]
tickets[1].bookings.extend(booking2)
person.tickets.extend(tickets)
session.add(person)
session.commit()
def query():
import json
with Session() as session:
session = Session()
res = session.query(Person).one()
print(json.dumps(res.dict(), indent=2))
res = session.query(Ticket).filter(Ticket.ticket_id == "10001000").one()
print(json.dumps(res.dict(), indent=2))
print(json.dumps(res.persons.dict(), indent=2))
res = session.query(Booking).filter(Booking.booking_id == "100010011000").one()
print(json.dumps(res.dict(), indent=2))
print(json.dumps(res.ticket_back_populates.dict(), indent=2))
print(json.dumps(res.ticket_back_populates.persons.dict(), indent=2))
if __name__ == "__main__":
insert_data()
query()
実行結果
# print(json.dumps(res.dict(), indent=2))
{
"id": "1",
"name": "hoge",
"tickets": [
{
"person_id": "1",
"ticket_id": "10001000",
"room_name": "hoge",
"bookings": [
{
"person_id": "1",
"ticket_id": "10001000",
"booking_id": "100010001000",
"comment": "first ticket"
}
]
},
{
"person_id": "1",
"ticket_id": "10001001",
"room_name": "hage",
"bookings": [
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011000",
"comment": "first ticket"
},
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011001",
"comment": "second ticket"
}
]
}
]
}
# print(json.dumps(res.dict(), indent=2))
{
"person_id": "1",
"ticket_id": "10001000",
"room_name": "hoge",
"bookings": [
{
"person_id": "1",
"ticket_id": "10001000",
"booking_id": "100010001000",
"comment": "first ticket"
}
]
}
# print(json.dumps(res.persons.dict(), indent=2))
{
"id": "1",
"name": "hoge",
"tickets": [
{
"person_id": "1",
"ticket_id": "10001000",
"room_name": "hoge",
"bookings": [
{
"person_id": "1",
"ticket_id": "10001000",
"booking_id": "100010001000",
"comment": "first ticket"
}
]
},
{
"person_id": "1",
"ticket_id": "10001001",
"room_name": "hage",
"bookings": [
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011000",
"comment": "first ticket"
},
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011001",
"comment": "second ticket"
}
]
}
]
}
# print(json.dumps(res.dict(), indent=2))
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011000",
"comment": "first ticket"
}
# print(json.dumps(res.ticket_back_populates.dict(), indent=2))
{
"person_id": "1",
"ticket_id": "10001001",
"room_name": "hage",
"bookings": [
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011000",
"comment": "first ticket"
},
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011001",
"comment": "second ticket"
}
]
}
# print(json.dumps(res.ticket_back_populates.persons.dict(), indent=2))
{
"id": "1",
"name": "hoge",
"tickets": [
{
"person_id": "1",
"ticket_id": "10001000",
"room_name": "hoge",
"bookings": [
{
"person_id": "1",
"ticket_id": "10001000",
"booking_id": "100010001000",
"comment": "first ticket"
}
]
},
{
"person_id": "1",
"ticket_id": "10001001",
"room_name": "hage",
"bookings": [
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011000",
"comment": "first ticket"
},
{
"person_id": "1",
"ticket_id": "10001001",
"booking_id": "100010011001",
"comment": "second ticket"
}
]
}
]
}
おまけ
構築したいDBのサンプルにuniqを追加したパターン
python(全文)
from sqlalchemy import VARCHAR, Column, ForeignKey, ForeignKeyConstraint, MetaData, UniqueConstraint, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, scoped_session, sessionmaker
url = "sqlite:///"
engine = create_engine(url, echo=False)
Session = scoped_session(sessionmaker(bind=engine, autocommit=False, autoflush=False))
Base = declarative_base()
class Person(Base):
__tablename__ = "persons"
id = Column(VARCHAR(4), autoincrement=False, primary_key=True)
name = Column(VARCHAR(30), nullable=False)
tickets = relationship("Ticket", backref="persons")
def dict(self):
tickets = [ticket.dict() for ticket in self.tickets]
res = {"id": self.id, "name": self.name, "tickets": tickets}
return res
class Ticket(Base):
__tablename__ = "tickets"
__table_args__ = (UniqueConstraint("person_id", "ticket_id", name="uniq_person_id_ticket_id"),)
person_id = Column(VARCHAR(4), ForeignKey("persons.id"), primary_key=True, autoincrement=False)
ticket_id = Column(VARCHAR(8), primary_key=True, autoincrement=False)
room_name = Column(VARCHAR(30), nullable=False)
bookings = relationship(
"Booking", foreign_keys="[Booking.person_id, Booking.ticket_id]", back_populates="ticket_back_populates"
)
def dict(self):
bookings = [booking.dict() for booking in self.bookings]
res = {
"person_id": self.person_id,
"ticket_id": self.ticket_id,
"room_name": self.room_name,
"bookings": bookings,
}
return res
class Booking(Base):
__tablename__ = "bookings"
__table_args__ = (
UniqueConstraint("person_id", "ticket_id", "booking_id", name="uniq_person_id_ticket_id_booking_id"),
ForeignKeyConstraint(["person_id", "ticket_id"], ["tickets.person_id", "tickets.ticket_id"]),
)
person_id = Column(VARCHAR(4), primary_key=True, autoincrement=False)
ticket_id = Column(VARCHAR(8), primary_key=True, autoincrement=False)
booking_id = Column(VARCHAR(12), primary_key=True, autoincrement=False)
comment = Column(VARCHAR(30), nullable=True)
ticket_back_populates = relationship(
"Ticket", foreign_keys="[Booking.person_id, Booking.ticket_id]", back_populates="bookings"
)
def dict(self):
res = {
"person_id": self.person_id,
"ticket_id": self.ticket_id,
"booking_id": self.booking_id,
"comment": self.comment,
}
return res
metadata = MetaData()
metadata.reflect(engine)
Base.metadata.create_all(bind=engine, checkfirst=True)
def insert_data():
with Session() as session:
person = Person(id=1, name="hoge")
tickets = [
Ticket(person_id="1000", ticket_id="10001000", room_name="hoge"),
Ticket(person_id="1000", ticket_id="10001001", room_name="hage"),
]
booking1 = Booking(person_id="1000", ticket_id="10001000", booking_id="100010001000", comment="first ticket")
tickets[0].bookings.append(booking1)
booking2 = [
Booking(person_id="1000", ticket_id="10001001", booking_id="100010011000", comment="first ticket"),
Booking(person_id="1000", ticket_id="10001001", booking_id="100010011001", comment="second ticket"),
]
tickets[1].bookings.extend(booking2)
person.tickets.extend(tickets)
session.add(person)
session.commit()
def query():
import json
with Session() as session:
session = Session()
res = session.query(Person).one()
print(json.dumps(res.dict(), indent=2))
res = session.query(Ticket).filter(Ticket.ticket_id == "10001000").one()
print(json.dumps(res.dict(), indent=2))
print(json.dumps(res.persons.dict(), indent=2))
res = session.query(Booking).filter(Booking.booking_id == "100010011000").one()
print(json.dumps(res.dict(), indent=2))
print(json.dumps(res.ticket_back_populates.dict(), indent=2))
print(json.dumps(res.ticket_back_populates.persons.dict(), indent=2))
if __name__ == "__main__":
insert_data()
query()
