ファイル連携のバッチを、pythonでDDDの手法をとりいれた実装例
背景
諸々の事情で、処理に必要なデータをもっているDBに直接アクセスすることができず、ファイルで連携してもらうケースがあります。その場合、処理の結果も、ファイルを出力する形で、別のシステムと連携することが多いです。
ファイルベースでの連携自体は、昔からよくある手法ではあるものの、手続き型よりの実装になりがちです。そのため、「どこからファイルをもらうのか」「どのように加工するのか」「作ったものをどこに連携するのか」、といったことが、新規メンバーが読みとりづらく、保守・運用の改修で、踏襲すべき手順をとりこぼしてしまって、（軽微な）バグ・デグレが発生しがちです。
そこで、手続き型よりの実装から、DDDのように処理に出てくる対象に焦点をあてて、もっとオブジェクト指向よりなコードに落とし込むことを試みてみました。ファイル連携のバッチの自分の中の暗黙知を、コードとして形式知化するといってもいいです。コードを読む能力が要求されますが、これによって新規メンバーも仕様を把握しやすくなり、安全かつ早く改修できるようになるはずです。
「現場で役立つシステム設計の原則 〜変更を楽で安全にするオブジェクト指向の実践技法」を読んだことが、大きな動機づけになっています。
pythonを使っているのは、前身となるプロジェクトがかなり単純なバッチ処理で、Javaを使うほどではなくpythonでやったから、後続のプロジェクトもそれに習った、という感じです。
ファイル連携のメリット・デメリット
やや脱線ですが、ファイル連携のメリット・デメリットを整理しておきます。
デメリットの多くは、実装を工夫することが概ね回避できます。（逆にいうと、知らずにやると、問題に直面します。）
メリット
- 連携先のシステムとは疎結合になる
- ファイルを出したあとであれば、連携先のシステムがダウンしても処理を開始・継続できる
- 開発環境には、連携先のシステム相当のサーバ、モックなどが不要
- ファイルを所定の場所に置くだけで動かせる
- 5分毎など、定期的にファイルをチェックしにいく形をとれば、連携されるタイミングが不定の場合にも対応できる
デメリット
- ファイルを置く側・回収する側の、それぞれの責務を明確にしないと、トラブルが起きやすい
- 例
- 連携先にファイルが残ったままでディスクフルになる
- 作成中のファイルをとりこんでしまう
- 例
- 改行コード、文字コードなどのファイルの仕様を明確に決めないと、処理に入る前でつまずく
- ファイルサイズが大きい場合、ファイルの移動だけで時間がかかる
- 本番作業が長時間化しやすい
- ファイルサイズが大きい場合、工夫しないと、ディスクやメモリがあふれやすい
- gzip圧縮したまま扱う
- （安易な）コードより、osコマンドの方が効率がいいケースがある
処理の概要
購入金額がある注文履歴データと、エントリーデータを受け取って、エントリーした人に、購入金額の1%のポイントをつける、というものです。
処理としてある程度の複雑さをもたせるために、連携先とはフラットなcsvでやりとりし、自分たちではそれをgzip圧縮した状態で処理する形をとっています。実際は、gzip圧縮した状態でファイルをもらうのがベストです。諸々の事情でそれができない場合は、自分たちのサーバに移すときに圧縮しながらコピーするとよいです。
ファイルを置く側
- 連携先に処理に必要なファイル(入力ファイル)を置く
- ファイル配置の完了を知らせるための、トリガーファイルを置く
ファイルを回収する側（バッチとして実装する部分）
- (cronなどで)定期的に、連携先にトリガーファイルがあるかをチェック
- トリガーファイルがなければなにもしない
- トリガーファイルがあれば処理を開始
4. トリガーファイルを削除
5. 入力ファイルをgzip圧縮して、作業用ディレクトリに移動
6. 入力ファイルをもとに、結果ファイルを作業用ディレクトリ上で、gzip圧縮した状態で作成
7. 連携先に、結果ファイルを解凍しながらコピー
8. 入力ファイル・結果ファイルをバックアップ用ディレクトリに移動
9. 作業用ディレクトリを削除
サンプルコード
パッケージ構成
ほぼ、Javaと同じノリでやっています。pythonのファイルは、モジュール扱いなので、Javaと違って、1クラス1ファイルにする必要はないはずです。
ただ、ファイルの一覧から、仕様がとらえやすくなるのと、git管理で差分が見やすくなりそうなので、基本、1クラス1ファイルでやってみました。
└── src
├── app_name.py
├── batch
│ ├── batch.py
│ ├── file_base_batch.py
│ └── file_lock.py
├── decorator
│ └── logging.py
├── domain
│ ├── backup
│ │ └── backup_directory.py
│ ├── input
│ │ ├── entry_file.py
│ │ ├── input_directory.py
│ │ ├── input_file.py
│ │ └── order_history_file.py
│ ├── output
│ │ ├── output_directory.py
│ │ └── result_file.py
│ └── work
│ ├── work_directory.py
│ ├── work_entry_file.py
│ ├── work_input_file.py
│ └── work_order_history_file.py
├── logging.yaml
├── main.py
├── mixin
│ └── direcotry
│ ├── existence_assured_directory.py
│ └── removable_directory.py
└── setup.py
| パッケージ名 | 概要 |
|---|---|
| batch | ロックファイルによる排他制御を伴うバッチを表すクラスの集まり |
| decorator | メソッドの経過時間をとるなどのデコレータの集まり |
| domain | 処理の中心となる、ファイル・ディレクトリを表したクラスの集まり |
| mixin | ファイル・ディレクトリの性質を表すための、trait、抽象クラスの集まり |
ファイル連携のバッチにおけるドメイン
現時点では、「処理に使われるファイル」「それらが置かれているディレクトリ」を焦点にあてるとうまくいきそうです。これらをクラスとして切り出すと、ファイルの動きを表現しやすくなります。
- 入力用ディレクトリ: InputDirectory
- 入力用ファイル: InputFile
というように、クラスを切り出していきます。（クラスを切り出せたからこそ、上記のような言葉が出せている面もあります。）
ディレクトリの移動は下記のようになります。
入力ファイル -> 作業用ファイル -> 結果ファイルの関係も、下記のようにクラスを切り出すと表現しやすいです。
一連の処理の実装は下記のようになります。
def execute(self):
input_directory = InputDirectory()
if not input_directory.trigger_file.exists():
logger.info(f'trigger file({input_directory.trigger_file}) does not exist. nothing to do.')
return
input_directory.trigger_file.unlink()
logger.info('start process since trigger file exists')
order_history_file = OrderHistoryFile(input_directory)
entry_file = EntryFile(input_directory)
input_files = [order_history_file, entry_file]
for input_file in input_files:
logger.info(f'checksum of {input_file.name}: {input_file.md5_checksum()}')
work_directory = WorkDirectory()
work_order_history_file = WorkOrderHistoryFile(order_history_file, work_directory)
work_entry_file = WorkEntryFile(entry_file, work_directory)
output_directory = OutputDirectory()
output_directory.trigger_file.unlink(missing_ok=True)
result_file = ResultFile(output_directory, work_directory)
result_file.create(work_order_history_file, work_entry_file)
output_directory.trigger_file.touch()
backup_directory = BackupDirectory()
work_directory.move_files_to(backup_directory)
work_directory.remove_if_exists()
ディレクトリの性質
タイミングはどうあれ、処理にでてくるディレクトリの下にファイルが置かれます。
そのため、「ディレクトリの存在が担保される」必要があります。Path.mkdirを使えばディレクトリの作成自体は簡単ですが、これの呼び出しを頻繁にやるのは厳しいです。背景を知らない人が改修した場合、ディレクトリ作成の部分は漏れやすいです。
そこで、「存在が担保されたディレクトリ」という抽象クラスを導入することにしました。
このクラスのコンストラクタでディレクトリ作成をしています。なので、このクラスを継承すれば、特に意識しなくても、インスタンスを生成したタイミングでディレクトリが作成されます。
from abc import ABC
from pathlib import Path
class ExistenceAssuredDirectory(ABC):
def __init__(self, *segments: str):
self.__path = Path(*segments)
self.__path.mkdir(exist_ok=True, parents=True)
assert self.__path.exists()
assert self.__path.is_dir()
@property
def path(self):
return self.__path
今回のサンプルコードではやってないですが、「backup/{yyyymm}とバックアップ用ディレクトリを月次でローテション、１ヶ月前のはディレクトリごと削除」ということをする場合は、「削除できるディレクトリ」という抽象クラスを用意すると、実装しやすくなります。
保守・運用の観点から便利な機能
実践では、コードを編集しやすくするためや、安全にバッチを実行するための仕組みが求められます。
他にもいろいろあるのですが、サンプルコードには以下の機能をいれてあります。
- ロックファイルによる排他制御
2. 実行中に、別プロセスで処理が開始されないようにするため - ログ設定ファイルのyaml化
- アプリケーション用ロガー
4. 各ファイルのロガーをアプリケーション用ロガーの子ロガーとすることで、クラス・パッケージの追加・編集をしても、ログ設定ファイルの編集は原則不要にできる - メソッドの経過時間をログ出力するためのデコレーター
まとめ
DDDの手法はそれなりの規模（マネタイズできている or 重要な業務を担っている）のwebアプリでないと、なかなかできないと思います。また、多くの書籍は、webアプリを想定したような形になっていると思います。
バッチ処理に対して実践するのは、どこからどのようにクラスを切り出せばいいのか、産みの苦しみがありました。そのおかけで、自分の理解をより深めることができ、pythonの使い方のレベルも上げられました。
バッチ処理の場合は、事業側を巻き込まずとも、ユビキタス言語を作ろうと思えば作れるので、その点ではDDDの手法を試しやすいともいえます。
正直、Javaのような静的型付け言語の方が、今回やろうとしていることには向いている気がします。
PythonにもGenericsがあるもののも、Javaに比べると書きづらい感があります。しばらくはやり続ける必要がありそうなので、うまくいったパターンをまとめた本をいずれは作ろうと思います。
Discussion