Pythonあれこれ（Vol.87掲載）

著者：飯尾淳

本連載では「Pythonを昔から使っているものの、それほど使いこなしてはいない」という筆者が、いろいろな日常業務をPythonで処理することで、立派な「蛇使い」に育つことを目指します。その過程を温かく見守ってください。皆さんと共に勉強していきましょう。第17回では、データを記録して残す「永続化」の手法の続編として、O/Rマッパーを使ってデータベースにアクセスする方法について解説します。

シェルスクリプトマガジン Vol.87は以下のリンク先でご購入できます。

図3　宣言的マッピングをするコード

from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
  pass

class User(Base):
  __tablename__ = "user_account"
  id: Mapped[int] = mapped_column(primary_key=True)
  name: Mapped[str] = mapped_column(String(30))
  fullname: Mapped[Optional[str]]
  addresses: Mapped[List["Address"]] = relationship(
    back_populates="user", cascade="all, delete-orphan"
  )
  def __repr__(self) -> str:
    return f"User(id={self.id!r},\
name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
  __tablename__ = "address"
  id: Mapped[int] = mapped_column(primary_key=True)
  email_address: Mapped[str]
  user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
  user: Mapped["User"] = relationship(back_populates="addresses")
  def __repr__(self) -> str:
    return f"Address(id={self.id!r},\
email_address={self.email_address!r})"

from typing import List

from typing import Optional

from sqlalchemy import ForeignKey

from sqlalchemy import String

from sqlalchemy.orm import DeclarativeBase

from sqlalchemy.orm import Mapped

from sqlalchemy.orm import mapped_column

from sqlalchemy.orm import relationship

class Base(DeclarativeBase):

pass

class User(Base):

__tablename__ = "user_account"

id: Mapped[int] = mapped_column(primary_key=True)

name: Mapped[str] = mapped_column(String(30))

fullname: Mapped[Optional[str]]

addresses: Mapped[List["Address"]] = relationship(

back_populates="user", cascade="all, delete-orphan"

)

def __repr__(self) -> str:

return f"User(id={self.id!r},\

name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):

__tablename__ = "address"

id: Mapped[int] = mapped_column(primary_key=True)

email_address: Mapped[str]

user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))

user: Mapped["User"] = relationship(back_populates="addresses")

def __repr__(self) -> str:

return f"Address(id={self.id!r},\

email_address={self.email_address!r})"

図4　データベースエンジンとなるオブジェクトを作成するコード

from sqlalchemy import create_engine
engine = create_engine("sqlite:///a_db.sqlite3", echo=True)

1 2	from sqlalchemy import create_engine engine = create_engine("sqlite:///a_db.sqlite3", echo=True)

図5　発行されるSQL

CREATE TABLE user_account (
	id INTEGER NOT NULL, 
	name VARCHAR(30) NOT NULL, 
	fullname VARCHAR, 
	PRIMARY KEY (id)
)
CREATE TABLE address (
	id INTEGER NOT NULL, 
	email_address VARCHAR NOT NULL, 
	user_id INTEGER NOT NULL, 
	PRIMARY KEY (id), 
	FOREIGN KEY(user_id) REFERENCES user_account (id)
)

CREATE TABLE user_account (

id INTEGER NOT NULL,

name VARCHAR(30) NOT NULL,

fullname VARCHAR,

PRIMARY KEY (id)

)

CREATE TABLE address (

id INTEGER NOT NULL,

email_address VARCHAR NOT NULL,

user_id INTEGER NOT NULL,

PRIMARY KEY (id),

FOREIGN KEY(user_id) REFERENCES user_account (id)

)

図6　データ操作用オブジェクトを作成するコード

from sqlalchemy.orm import Session

with Session(engine) as session:
  spongebob = User(
    name="spongebob",
    fullname="Spongebob Squarepants",
    addresses=[Address(email_address="spongebob@sqlalchemy.org")],
  )
  sandy = User(
    name="sandy",
    fullname="Sandy Cheeks",
    addresses=[
      Address(email_address="sandy@sqlalchemy.org"),
      Address(email_address="sandy@squirrelpower.org"),
    ],
  )
  patrick = User(name="patrick", fullname="Patrick Star")
  session.add_all([spongebob, sandy, patrick])
  session.commit()

from sqlalchemy.orm import Session

with Session(engine) as session:

spongebob = User(

name="spongebob",

fullname="Spongebob Squarepants",

addresses=[Address(email_address="spongebob@sqlalchemy.org")],

)

sandy = User(

name="sandy",

fullname="Sandy Cheeks",

addresses=[

Address(email_address="sandy@sqlalchemy.org"),

Address(email_address="sandy@squirrelpower.org"),

)

patrick = User(name="patrick", fullname="Patrick Star")

session.add_all([spongebob, sandy, patrick])

session.commit()

図7　図6のコードを実行した際の画面出力（抜粋）

BEGIN (implicit)
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
（略）('spongebob', 'Spongebob Squarepants')
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
（略）('sandy', 'Sandy Cheeks')
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
（略）('patrick', 'Patrick Star')
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
（略）('spongebob@sqlalchemy.org', 1)
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
（略）('sandy@sqlalchemy.org', 2)
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
（略）('sandy@squirrelpower.org', 2)
COMMIT

BEGIN (implicit)

INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id

（略）('spongebob', 'Spongebob Squarepants')

INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id

（略）('sandy', 'Sandy Cheeks')

INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id

（略）('patrick', 'Patrick Star')

INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id

（略）('spongebob@sqlalchemy.org', 1)

INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id

（略）('sandy@sqlalchemy.org', 2)

INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id

（略）('sandy@squirrelpower.org', 2)

COMMIT

図8　簡単な検索をするコード

from sqlalchemy.orm import Session
from sqlalchemy import select

session = Session(engine)
stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))
for user in session.scalars(stmt):
  print(user)

from sqlalchemy.orm import Session

from sqlalchemy import select

session = Session(engine)

stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))

for user in session.scalars(stmt):

print(user)

図9　図8のコードを実行した際に発行されるSQL

BEGIN (implicit)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name IN (?, ?)
（略） ('spongebob', 'sandy')

BEGIN (implicit)

SELECT user_account.id, user_account.name, user_account.fullname

FROM user_account

WHERE user_account.name IN (?, ?)

（略） ('spongebob', 'sandy')

図10　図8のコードを実行した際に表示される検索結果

User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')

1 2	User(id=1, name='spongebob', fullname='Spongebob Squarepants') User(id=2, name='sandy', fullname='Sandy Cheeks')

図11　少し複雑な検索をするコード

stmt = (
  select(Address)
  .join(Address.user)
  .where(User.name == "sandy")
  .where(Address.email_address == "sandy@sqlalchemy.org")
)
sandy_address = session.scalars(stmt).one()

stmt = (

select(Address)

.join(Address.user)

.where(User.name == "sandy")

.where(Address.email_address == "sandy@sqlalchemy.org")

)

sandy_address = session.scalars(stmt).one()

図12　図11のコードを実行した際に発行されるSQL

SELECT address.id, address.email_address, address.user_id
FROM address JOIN user_account ON user_account.id = address.user_id
WHERE user_account.name = ? AND address.email_address = ?
（略） ('sandy', 'sandy@sqlalchemy.org')

SELECT address.id, address.email_address, address.user_id

FROM address JOIN user_account ON user_account.id = address.user_id

WHERE user_account.name = ? AND address.email_address = ?

（略） ('sandy', 'sandy@sqlalchemy.org')

図13　データ変更のシンプルなコード例

sandy_address.email_address = "sandy_cheeks@sqlalchemy.org"
session.commit()

1 2	sandy_address.email_address = "sandy_cheeks@sqlalchemy.org" session.commit()

図14　図13のコードを実行した際に発行されるSQL

UPDATE address SET email_address=? WHERE address.id = ?
（略）('sandy_cheeks@sqlalchemy.org', 2)
COMMIT

UPDATE address SET email_address=? WHERE address.id = ?

（略）('sandy_cheeks@sqlalchemy.org', 2)

COMMIT

図15　データ変更の少し複雑なコード例

stmt = select(User).where(User.name == "patrick")
patrick = session.scalars(stmt).one()
patrick.addresses \
.append(Address(email_address="patrickstar@sqlalchemy.org"))
session.commit()

stmt = select(User).where(User.name == "patrick")

patrick = session.scalars(stmt).one()

patrick.addresses \

.append(Address(email_address="patrickstar@sqlalchemy.org"))

session.commit()

図16　図15のコードを実行した際に発行されるSQL


SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = ?
（略）('patrick',) 
　
SELECT address.id AS address_id, address.email_address
  AS address_email_address, address.user_id AS address_user_id 
FROM address 
WHERE ? = address.user_id
（略）(3,)
　
INSERT INTO address (email_address, user_id) VALUES (?, ?)
（略）('patrickstar@sqlalchemy.org', 3)
COMMIT

SELECT user_account.id, user_account.name, user_account.fullname

FROM user_account

WHERE user_account.name = ?

（略）('patrick',)

SELECT address.id AS address_id, address.email_address

AS address_email_address, address.user_id AS address_user_id

FROM address

WHERE ? = address.user_id

（略）(3,)

INSERT INTO address (email_address, user_id) VALUES (?, ?)

（略）('patrickstar@sqlalchemy.org', 3)

COMMIT