Pythonでの実務でよくありそうな内容のリファクタを検討してみた

概要

ORMを使ったレコードの操作は便利ですが、特に更新操作時に気をつけないとスパゲッティコードの種になりがちと思ったことと、実際書かれがちなコードを改善していこうと思います。

実装

下記は簡単な例です。
<User(name='Yamada Taro', age='40')>になるかと思います。

from sqlalchemy import Column, Integer, Sequence, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

engine = create_engine("sqlite:///:memory:", echo=False)

Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    def __repr__(self):
        return "<User(name='%s', age='%s')>" % (self.name, self.age)

    id = Column(Integer, Sequence("user_id_seq"), primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

new_user = User(name="John Doe", age=30)
session.add(new_user)
session.commit()

user = session.query(User).filter_by(name="John Doe").first()
user.name = "Yamada Taro"
user.age = 40
session.commit()

print(user)

単純なスクリプトであればこれでいいのですが、実際の業務コードでは様々なロジックが行き交います。
当初の要件が変わったものとして、かわいいペットも増やし、実際筆者が遭遇したことのあるコードに近づけたいと思います。

from sqlalchemy import Column, Integer, Sequence, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

engine = create_engine("sqlite:///:memory:", echo=False)

Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    def __repr__(self):
        return "<User(name='%s', age='%s')>" % (self.name, self.age)

    id = Column(Integer, Sequence("user_id_seq"), primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


class Pet(Base):
    __tablename__ = "pets"

    def __repr__(self):
        return "<Pet(name='%s', age='%s')>" % (self.name, self.age)

    id = Column(Integer, Sequence("pet_id_seq"), primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


def update_name(user: User, name: str) -> None:
    user.name = name


def create_pet(session: Session, name: str, age: int) -> None:
    pet = Pet(name=name, age=age)
    session.add(pet)

    try:
        session.commit()
    except Exception as e:
        session.rollback()
        print(e)


def update_age(user: User, age: int) -> None:
    user.age = age


Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

new_user = User(name="John Doe", age=30)
session.add(new_user)
session.commit()

user = session.query(User).filter_by(name="John Doe").first()

# 値の集約化として値を格納しておく
user_dict = {}
user_dict.setdefault("name", "Yamada Taro")
user_dict.setdefault("age", 40)

per_dict = {}
per_dict.setdefault("name", "Tama")
per_dict.setdefault("age", 5)

update_name(user, user_dict["name"])
create_pet(session, per_dict["name"], per_dict["age"])
update_age(user, user_dict["age"])
try:
    session.commit()
except Exception as e:
    session.rollback()
    print(e)

print(user)

同様に<User(name='Yamada Taro', age='40')>になるかと思いますが、バグになりそうな点があります。

• create_petでコミットしてしまっており、もし後続のコミットで失敗した場合、age=30のYamada Taroができてしまう
• 条件分岐や処理が増えていくと、userモデルが引き回されている中で、いつどこで何をsetされているのかがわかりにくくなる
• user_dictとper_dictはdictである以上の情報を得ることができないことと、dictはどこからでもsetできてしまうので、いつどこで内容が変わるのかがわからない不安がつきまとう

なので、個人的にリファクタするときは、

• ValueObjectで属性値を管理しつつイミュータブルにする
• モデルへの値のsetは別途作成したインターフェースを経由するものとする
• モデルを引き回さずに、上記で作成したValueObjectを処理中で作成し、モデルを触るときは永続化直前にする
• メソッド・関数中は見えづらいので、commit()が明示的に必要な部分は本処理中に書く。

改善例は下記です。

from dataclasses import dataclass

from sqlalchemy import Column, Integer, Sequence, String, create_engine
from sqlalchemy.orm import Session, declarative_base, sessionmaker

engine = create_engine("sqlite:///:memory:", echo=False)

Base = declarative_base()


@dataclass(frozen=True)
class UserUpdatesData:
    name: str
    age: int


class User(Base):
    __tablename__ = "users"

    def __repr__(self):
        return "<User(name='%s', age='%s')>" % (self.name, self.age)

    id = Column(Integer, Sequence("user_id_seq"), primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


class UserUpdator:
    def __init__(self, session: Session, user: User):
        self.session = session
        self.user = user

    def exec(self, data: UserUpdatesData) -> None:
        self.user.name = data.name
        self.user.age = data.age


@dataclass(frozen=True)
class PetCreatesData:
    name: str
    age: int


class Pet(Base):
    __tablename__ = "pets"

    def __repr__(self):
        return "<Pet(name='%s', age='%s')>" % (self.name, self.age)

    id = Column(Integer, Sequence("pet_id_seq"), primary_key=True)
    name = Column(String(50))
    age = Column(Integer)


class PetCreator:
    def __init__(self, session: Session):
        self.session = session

    def exec(self, data: PetCreatesData) -> Pet:
        p = Pet(name=data.name, age=data.age)
        self.session.add(p)


Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

new_user = User(name="John Doe", age=30)
session.add(new_user)
session.commit()

# ユーザーの更新
u = UserUpdatesData(name="Yamada Taro", age=40)
user = session.query(User).filter_by(name="John Doe").first()
UserUpdator(session, user).exec(u)

# ペットの作成
p = PetCreatesData(name="Tama", age=5)
PetCreator(session).exec(p)
try:
    session.commit()
except Exception as e:
    session.rollback()
    print(e)

print(user)

良くなったと思うことは下記です。

• UserUpdatesData/PetCreatesDataに値を集約および、値の再設定を許さない形にできた
• ルール化は必要ですが、値の更新やレコードを追加する操作を別クラスに切り出すことで集約化と見通しが良くなった
• userモデルがコード中に出てて値更新で連れ回すことが無くなった
• dataclassで定義しているのでエディタの参照しやすく、補完が効きやすくなった。型チェッカーを入れることで不整合な値を検知することもできる
  
  

まだまだ改善点はあると思いますが、以上です。
ありがとうございました。