skydum

個人的な作業記録とか備忘録代わりのメモ

SQLAlchemyで外部制約キーを複数持ったリレーションシップの設定

SQLAlchemyを使って外部制約キーを複数持ったリレーションシップを設定する

リレーションシップの基本

構築したいDBのER図

ER図

生成したいSQL

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図

ER図基本パターン

python

  • 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図(最初に貼ったのと同じ)

ER図(最初と同じ)

python

  • 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"
        }
      ]
    }
  ]
}

おまけ

  • 複数のカラムにuniq制約をつける

  • sqlalchemyのテーブル定義tips qiita.com

構築したい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()