peketamin/generate_user_list.md

## generate_user_list.md

      
    Raw
  

              generate_user_list.md
            
          
    ユーザーレコードをDBから読み込んでCSVを出力するクラスを作る、というお題に対して、どうオブジェクト指向で実装するか (Python)

手続き指向での提出例

import datetime
from typing import Dict, List

import pandas as pd

from django.core.validators import validate_email
from django.db import connection


def dictfetchall(cursor) -> List[Dict]:
    """Return all rows from a cursor as a dict.

    see: https://docs.djangoproject.com/en/1.11/topics/db/sql/#executing-custom-sql-directly
    """
    columns = [col[0] for col in cursor.description]
    return [
        dict(zip(columns, row))
        for row in cursor.fetchall()
    ]


class MissingDataError(Exception):
    """If data is empty or something like that."""


class UserListGenerater:
    DATE_FORMAT = "%Y-%m-%d"

    COL_ID = "id"
    COL_NAME = "name"
    COL_EMAIL = "email"
    COLUMNS = [COL_ID, COL_NAME, COL_EMAIL]

    QUERY = """
        SELECT
           id
          ,name
          ,email
        FROM
          users
          INNER JOIN
            user_actions ON (user_actions.user_id = users.id)
        WHERE
          users.campaign_id = %(campaing_id)s
          AND
          user_actions.date = %(date)s
        GROUP BY
           name
          ,email
    """

    def __init__(self, campaign_id: int, target_date: datetime.date) -> None:
        self.campaign_id = campaign_id
        self.target_date = target_date.strftime(self.DATE_FORMAT)

        self._cur = connection.cursor()

    def generate_csv(self) -> bytes:
        return self._generate()._as_csv()

    def generate_df(self) -> pd.DataFrame:
        return self._generate()._as_df()

    @property
    def _params(self):
        params = {"campaign_id": self.campaign_id, "date": self.target_date}
        return params

    def _execute_query(self):
        self._cur.execute(self.QUERY, self._params)
        return dictfetchall(self._cur)

    def _generate(self):
        users = self._execute_query()
        df = pd.DataFrame(users)
        if df.empty:
            raise MissingDataError(
                f"No user for campaign id {self.campaign_id}, target date: {self.target_date}"
            )

        self.df = df
        self._validate()

        return self

    def _as_df(self):
        return self.df

    def _as_csv(self):
        csv = self.df.loc[:, self.COLUMNS].to_csv(index=False)
        csv_b = csv.encode(self.FILE_ENCODING)
        return csv_b

    ## VALIDATION  LOGIC ######################################################
    def _validate(self):
        # Hook here other validation logic.
        self._validate_invalid_email()

    def _validate_invalid_email(self):
        for email in self.df[self.COL_EMAIL].to_numpy():
            validate_email(email)
利用例
csv_b = UserListGenerater(campaign_id=3, target_date=datetime.date(2019, 7, 1)).generate_csv()
Good


GROUP BY をやる場合、Django ORM だとあまり直感的に書けないので、生 SQL を使うのは保守性面でもありだと思う。

ただし、開発時には忘れずにクエリログを出しながら、 ORM が吐くクエリが非効率なクエリになってないか、想定してないクエリになってないか、を確認すること
加えて、個々の SQL は Explain を取ってパフォーマンスに問題なさそうかを確認すること


private メソッドには先頭にアンダースコアをつけて public なメソッドと区別していること

おもうところ 🤔


cursor は一度しか呼ばれてないのに、なぜ self._cur に入れているのか。

再利用を想定しているなら、具体的な場面が出てくるまで不要な実装をしないこと (YAGNI)
実装意図が読み取れない行を追加しないこと
cursorは一度中身を吸い出されると空になり、また、未然の場合、データがflushされてないので、再利用には注意する必要がある

なにかの処理途中でデータが吸い出されると、予期しないデータが返ってくる


メソッドが不要に多い。一度しか呼ばれないものは必ずしも関数に分けなくても良い。

責務の分割をするのは良いが、読み下して十分理解可能なコードになっていれば、必要性が発生するまで無理に分割しなくてよい。
なぜ分割しているかの意図が読めないから。


データがない時に例外を出しているが、出力データが Collection 相当のデータであれば、空を返す方が一貫性がある。

わかりやすい例で言えば、まさに SQL の結果は一般的に、対象レコードがない場合はエラーを返すのではなく、空リストを返すのであり、一つの見本である。
SQLでは campaign_id で絞り込んでいるので、該当キャンペーンを持つユーザーがいないことが本当に、運用上、想定外の事象なのかを熟慮したい


datetime.date を文字列に直して保持し直しているが、その必要はない

もし再利用時のパフォーマンスを考慮しているなら、その恩恵に比べて、意図が読みづらい方が問題である
また、変換元のオブジェクトを保持する方が、「オブジェクトの生成に必要だった変数はなんだったか」を情報の欠落なく保持できる


など思うところがあるので、リファクタリングする。
オブジェクトを意識したリファクタリング例

import datetime
from typing import Dict, List, Optional

import pandas as pd

from django.core.validators import ValidationError, validate_email
from django.db import connection

# ...


class UserListContainer:
    QUERY = """
        SELECT
           id
          ,name
          ,email
        FROM
          users
          INNER JOIN
            user_actions ON (user_actions.user_id = users.id)
        WHERE
          users.campaign_id = %(campaing_id)s
          AND
          user_actions.date = %(date)s
        GROUP BY
           name
          ,email
    """
    COLUMNS = ['id', 'name', 'email']

    def __init__(self, campaign_id: int, target_date: datetime.date) -> None:
        self.campaign_id = campaign_id
        self.target_date = target_date
        self.df_users: pd.DataFrame = pd.DataFrame()

    @classmethod
    def factory(cls, campaign_id: int, target_date: datetime.date) -> 'UserListContainer':
        user_list_container= cls(campaign_id, target_date)
        cur = connection.cursor()
        cur.execute(self.QUERY, {
            "campaign_id": campaign_id,
            "date": self.target_date.strftime("%Y-%m-%d")
        })
        rows = dictfetchall(cur)
        df = pd.DataFrame(rows)
        user_list_container.df_users = df
        return user_list_container

    def validate_emails(self):
        err_emails: List[str] = []
        for email in self.df_users['email'].values:
            try:
                validate_email(email)
            except ValidationError:
                err_emails.append(email)
        return err_emails

    def to_csv(self) -> bytes:
        if self.df_users.empty:
            return pd.DataFrame(columns=self.COLUMNS).to_csv(index=False)
        return self.df_users.loc[:, self.COLUMNS].to_csv(index=False)
user_list_container = UserListContainer.factory(campaign_id=3, target_date=datetime.date(2019, 7, 1))
err_emails = user_list_container.validate_emails()
if not err_emails:
    csv_b = user_list_container.to_csv().encode('utf-8')
このオブジェクトを利用する側のテストでは、適当なオブジェクトが作りやすいし、モックもしやすくなる。
単純に
user_list_container = UserListContainer(
    campaign_id=3,
    target_date=datetime.date(2019, 7, 1),
)
user_list_container.df_users = pd.DataFrame({'id': [1], 'name': ['test'], 'email': ['e@e.com']})
というようにオブジェクトを作るだけで済むからである。
また、 user_list_container.df_users を参照することができるので len(user_list_container.df_users.index) を出力することも可能になった。
このような「生成済みプロパティ」へのアクセスは、元々の手続き指向版では出来ない。
また、メソッド数を減らし、変数も減らしている。「可能な限りシンプルに、ただし読みやすく」かつ、パフォーマンスに気をつけて、という感じ。
とにかくシンプルにしたい。
ただ、プログラミング初心者にとっては「シンプルに書く」難しいことかもしれない。
自分の中に「シンプルなもの」「シンプルなものが産むメリットの体験」「そうでないもの」「そうでないものが産むデメリットの体験」の量が少ないからだ。
少なくとも私はそうだった。
それなりの読み・書き演習量が必要。
添削を通して、小さな失敗を早く繰り替えして覚えていきたい。
心の師匠たちへの謝辞


https://twitter.com/mosa_siru :

「とにかくシンプルにして。書き下しでも読みやすければ、それはそれでアリです」
「それムズくね？原理主義過ぎてもダメです」
「三大原則: (ネストは浅く、副作用なく、命名を適切に) を必ず意識してください」
「ORM使うなら必ずクエリの実行計画見て」


https://twitter.com/_nishigori :

「オブジェクト指向を理解した方がいいです」この発言を聞いて「オブジェクト指向エクササイズ」を発見できた。


ねぎしさん

「別に int や float まで wrap することはないですけど、「オブジェクトのプロパティ」で代替可能な dict は専用 class を作ってもいいです。
DDDで言う "値オブジェクト", "エンティティ" などがそうです。dict のキーは想定外のものも入りやすいし、静的解析も不向きです (コード補完できない)。


https://teratail.com/questions/80635 の pashango2 氏コメントから引用


なるほどです。
すごくスッキリしていますが、これってOOP的に問題はないのですか？
pashango2
問題はありますよ、もしかしたらボロカスに言われるかもしれません。
しかし、この方法は「これ以上ないくらいシンプル」という何にも代えがたい力があります。
この方法で将来的に問題が出ないかもしれないのに、先回りしてシンプルさを崩す不可逆な変更をする事は愚策であると思います、不確定な未来に対しての過剰投資であり、早すぎる最適化です。

「すべてのプリミティブ型と文字列型をラップすること」

「ThoughtWorksアンソロジー」という技術書で「オブジェクト指向エクササイズ」というセクションがあり、
Javaプログラマの間で話題になった時期がありました。私は Java 界隈ではなかったので、ネットで調べた範囲でしか当時の様子を知りません。。。
「すべて」とは思いませんが、この考え方は目から鱗でした。
例えば「イベント日」 event_date というものがあったとします。
これを datetime.date で持っていると次のようなことができません。
if event_date.is_weekday():
    print('Work hard.')
else:
    print('Take a rest.')
手続き指向で書き換えるとこうでしょうか。
if is_weekday(event_date):
    ...
else:
    ...
is_weekday() が「イベント日」というドメインにのみ関係するのであれば、凝集性のために集約・局所化した方が、
その他の関数やオブジェクトの定義と切り離せて見通しがよくなります。
私が出会った記事では確か、「金額オブジェクト」を定義していて、その説明がとても得心したのですが、残念ながらその記事を見つけることが出来ませんでした。
あの記事を書いてくださった方に心から感謝します。