育てるアーキテクチャ：戦い抜くPythonマイクロサービスの設計と進化戦略

PyCon2025JP, 09/27 DAY2,
育てるアーキテクチャ：戦い抜くPythonマイクロサービスの設計と進化戦略
https://2025.pycon.jp/ja/timetable/talk/CVUNWL

Transcript

1. 育てるアーキテクチャ 株式会社テレシー VPoE daisuke-sasaki（@fujidomoe） PyConJP 2025 2025.09.27 戦い抜くマイクロサービスの設計と進化戦略
2. 株式会社テレシー VPoE daisuke-sasaki 略歴 株式会社テレシー 開発部 部長/VPoE • 2020年にテレシーにJOIN、リードエンジニアとしてアプリケー ションとインフラのエンジニアリングを行う。 • 2024年からテレシー VPoEに就任。 役割/領域 #高校野球 #少年野球 management engineering Cloud infra Back Manag ement Front @fujidomoe
6. おしらせ
7. AGENDA 背景 02 Pythonエコシステムの特徴 03 実際に遭遇したアンチパターン 04 改善の旅 05 チームでの実践的な改善のプロセス 01
8. 突然ですが、皆さんの職場にエンジニアでない方が書いた Pythonコードはありませんか? 󰢧
9. とりあえず動くコード 😇
10. def process(data_df): jp_indices = data_df[data_df['region'] == 'JP'].index data_df.drop(data_df.index.difference(jp_indices), inplace=True) data_df['ctr'] = data_df['clicks'] / data_df['views'] low_perf_indices = data_df[data_df['ctr'] < 0.02].index data_df.drop(low_perf_indices, inplace=True) data_df['is_high_performer'] = True 存在保証のない列でフィルタ &中間データの生成 破壊的な操作 存在保証のない列を用いて演算
11. 本日のゴールです • 技術は「きっかけ」、文化が「ゴール」 • アーキテクチャを育てるのはチームです。
12. このトークを聞いてもらいたい人 • 前のスライドの全部のせ関数が、本番で動いているのを 想像して「ウッ」となった方 • 実践的な設計パターンやアンチパターンに興味がある方 • セルフマージ文化に課題を感じている方
13. • 開発体制(2022年頃) ◦ 研究者とエンジニアは別チーム ◦ それぞれのチームで複数のマイクロサービスを開発 ◦ 1マイクロサービス -> 1人が担当、全てセルフマージ 本日お話する舞台の背景
14. • 開発体制(2022年頃) ◦ 研究者とエンジニアは別チーム ◦ それぞれのチームで複数のマイクロサービスを開発 ◦ 1マイクロサービス -> 1人が担当、全てセルフマージ 本日お話する舞台の背景
15. • 開発体制(2022年頃) ◦ 研究者とエンジニアは別チーム ◦ それぞれのチームで複数のマイクロサービスを開発 ◦ 1マイクロサービス -> 1人が担当、全てセルフマージ 本日お話する舞台の背景
16. • 開発体制(2022年頃) ◦ 研究者とエンジニアは別チーム ◦ それぞれのチームで複数のマイクロサービスを開発 ◦ 1マイクロサービス -> 1人が担当、全てセルフマージ 本日お話する舞台の背景
17. 研究職の方から本番稼働中のETLバッチを引き継いだ。 なんでETLバッチのイニシャルを研究職が書いたの?
18. 研究職の方から本番稼働中のETLバッチを引き継いだ。 なんでETLバッチのイニシャルを研究職が書いたの?
19. 組織で当たり前が揃っていない。レビュー文化がない。 その弾みで、 「試行錯誤中相当のコードが本番で動いている」 これを出発点に、「戦い抜くPythonマイクロサービスの設計 と進化戦略」それを契機としてチームの文化を変えていった 話をします。
20. • リアーキテクチャは、まぁ、やるよね。 • コードレビューも、まぁ、やるよね。 • 大事なのは推進プロセス • 最も気を使ったのは組織のソフト面
21. • リアーキテクチャは、まぁ、やるよね。 • コードレビューも、まぁ、やるよね。 • 大事なのは推進プロセス • 最も気を使ったのは組織のソフト面
22. • 今日話すこと ◦ 引き継いだバッチアプリケーション ▪ アンチパターンの紹介と具体的な改善 ▪ チームでの実践的な改善のプロセス
23. • 今日話さないこと • 特定フレームワーク、特定ORMの詳細解説
24. AGENDA 背景 02 Pythonエコシステムの特徴 03 実際に遭遇したアンチパターン 04 改善の旅 05 チームでの実践的な改善のプロセス 01
25. Pythonエコシステムの特徴 研究職(notエンジニア)の方が書いたコードを引き継ぐケー スがある
26. 稀によくある光景 Jupyter Notebookから始まったコードが、ほとんどそのま ま本番バッチとして稼働 😇
27. 稀によくある光景 • 設計思想がバラバラ ◦ あるものは「スクリプトの延長」 ◦ あるものは「オブジェクト指向」 • テストコードは存在しない。
28. AGENDA 背景 02 Pythonエコシステムの特徴 03 実際に遭遇したアンチパターン 04 改善の旅 05 チームでの実践的な改善のプロセス 01
29. 前置き: 実際に稼働していたコードへの敬意 • 紹介するコードは実際に本番で動きお金を稼いでくれた コード 💸 • 稼いだくれたコードに感謝と敬意を持って話します
30. アンチパターン紹介 • Python動的型付けの落とし穴 ◦ kwargs編 ◦ Pandas DataFrame編 • アーキテクチャ不在 • グローバル関数羅列
31. アンチパターン紹介 • Python動的型付けの落とし穴 ◦ kwargs編 ◦ Pandas DataFrame編 • アーキテクチャ不在 • グローバル関数羅列
32. Pythonは動的型付け言語でさくっと動く
33. アプローチ次第でこれが罠になりがちです
34. この辞書には、何のキーが入ってるんだっけ 🤔
35. kwargsとは？ def create_profile(name, **options): profile = {"name": name} # デフォルト値を設定 profile["age"] = options.get("age", "未設定") profile["email"] = options.get("email", "未設定") profile["location"] = options.get("location", "未設定") return profile
36. kwargs編① def extract_data(user_id, **kwargs): start = kwargs.get('start_date', '2025-09-27') if kwargs.get('inclued_deleted'): pass
37. kwargs編① def extract_data(user_id, **kwargs): start = kwargs.get('start_date', '2025-09-27') if kwargs.get('inclued_deleted'): pass 😱s/inclued/include 😱常にFalseになるが気づかない
38. kwargs編② # extract.py def extract_data(user_id, **kwargs): start_date = kwargs.get("start_date", '2025-09-27') end_date = kwargs.get("end_date", datetime.now()) include_deleted = kwargs.get('include_deleted', False) batch_size = kwargs.get('batch_size', 1000) retry_count = kwargs.get('retry_count', 3) # さらに続く... # transform.py def transform_data(data, **kwargs): normalize = kwargs.get('normalize', True) remove_duplicates = kwargs.get('remove_duplicates', True) # まだまだ続く...
39. kwargs編② # extract.py def extract_data(user_id, **kwargs): start_date = kwargs.get("start_date", '2025-09-27') end_date = kwargs.get("end_date", datetime.now()) include_deleted = kwargs.get('include_deleted', False) batch_size = kwargs.get('batch_size', 1000) retry_count = kwargs.get('retry_count', 3) # さらに続く... 😱 どんなパラメータを渡せるの？ 😱 利用可能なオプション覚えておくの？
40. アンチパターン紹介 • Python動的型付けの落とし穴 ◦ kwargs編 ◦ Pandas DataFrame編 • アーキテクチャ不在 • グローバル関数羅列
41. Pandas DataFrameとは? Pythonで表形式のデータ操作 を柔軟に扱うためのライブラ リです。
42. Pandas DataFrameを利用した例 import pandas as pd from sqlalchemy import create_engine db_engine = create_engine("省略") # 1. SQLクエリを準備 sql_query = "SELECT user_id, product_name, price FROM sales_data WHERE price > 5000;" # 2. SQLを実行し、結果を直接DataFrameに読み込む df = pd.read_sql(sql_query, db_engine) # 3. 取得したDataFrameを表示 print(df) # 例：価格(price)が10000より大きいデータを抽出 high_price_df = df[df['price'] > 10000]
43. Pandas Dataframe編 def process(self, target_df: pd.DataFrame, source_df: pd.DataFrame) -> pd.DataFrame: time_total_df = (target_df.groupby("datehour")[["count"]] .sum().reset_index().rename(columns={"count":"total_count"}) ) prorating_target_df = pd.merge(target_df, time_total_df, on="datehour", how="left") prorating_target_df = pd.merge( prorating_target_df, source_df.rename(columns={"count": "source_count"}), on="datehour", how="outer", ) df = prorating_target_df[["datehour","prefecture","prorated"]].rename(columns={"prorated": "count"}) # この後手続き処理が100行ほど続く...
44. Pandas Dataframe編 def process(self, target_df: pd.DataFrame, source_df: pd.DataFrame) -> pd.DataFrame: time_total_df = (target_df.groupby("datehour")[["count"]] .sum().reset_index().rename(columns={"count":"total_count"}) ) prorating_target_df = pd.merge( prorating_target_df, source_df.rename(columns={"count": "source_count"}), on="datehour", how="outer", ) df = prorating_target_df[["datehour","prefecture","prorated"]].rename(columns={"prorated": "count"}) # この後手続き処理が100行ほど続く... 😇これはこれで気になる
45. Pandas Dataframe編 def process(self, target_df: pd.DataFrame, source_df: pd.DataFrame) -> pd.DataFrame: time_total_df = (target_df.groupby("datehour")[["count"]] .sum().reset_index().rename(columns={"count":"total_count"}) ) # この後手続き処理が100行ほど続く... 😱引数も戻り値もDataFrame
46. Pandas Dataframe編 time_total_df = (target_df.groupby("datehour")[["count"]] .sum().reset_index().rename(columns={"count":"total_count"}) ) prorating_target_df = pd.merge(target_df, time_total_df, on="datehour", how="left") prorating_target_df = pd.merge( prorating_target_df, source_df.rename(columns={"count": "source_count"}), on="datehour", how="outer", ) df = prorating_target_df[["datehour","prefecture","prorated"]].rename(columns={"prorated": "count"}) メソッドの中身全体として 😱 存在保証のないカラムにアクセス 😱 各操作がDataFrameの構造を変更
47. アンチパターン紹介 • Python動的型付けの落とし穴 ◦ kwargs編 ◦ Pandas DataFrame編 • アーキテクチャ不在 • グローバル関数羅列
48. アーキテクチャ不在（グローバル関数羅列） class RDBConnection: def __init__(self): self.con = pymysql.connect( "省略") def execute_read_query (self, query: str): return result def find_report_by_date_query (date: str) -> str: q = f"""SELECT report_id FROM ...""" return q def find_report_by_date (date: str) -> list[int]: db = RDBConnection() q = find_report_by_date_query (date) res = db.execute_read_query (q) return [int(r[0]) for r in res] def find_report_by_id_query (id: int) -> list[int]: # 省略 def find_report_by_id_query (id: int) -> str: # 省略 # find_report_by_dateと同じようなパターンが沢山存在する ....
49. アーキテクチャ不在（グローバル関数羅列） def find_report_by_date (date: str) -> list[int]: db = RDBConnection() q = find_report_by_date_query (date) res = db.execute_read_query (q) return [int(r[0]) for r in res] 毎回コネクション ビジネスロジック(データ変換)、SQL生成、データ入出力
50. アーキテクチャ不在（グローバル関数羅列） def find_report_by_date_query(date: str) -> str: q = f"""SELECT report_id FROM ...""" return q def find_report_by_date(date: str) -> list[int]: db = RDBConnection() q = find_report_by_date_query(date) res = db.execute_read_query(q) return [int(r[0]) for r in res] 必ずセット
51. AGENDA 背景 02 Pythonエコシステムの特徴 03 実際に遭遇したアンチパターン 04 改善の旅 05 チームでの実践的な改善のプロセス 01
52. Step by Stepアプローチ • Step1 : 型ヒントを追加してみる • Step2 : Pydanticで1つの関数の引数を整理 • Step3 : レイヤー分離を試す • Step4 : チーム全体で改善の輪をひろげる
53. Step by Stepアプローチ • Step1 : 型ヒントを追加してみる(割愛) • Step2 : Pydanticで1つの関数の引数を整理 • Step3 : レイヤー分離を試す • Step4 : チーム全体で改善の輪をひろげる
54. アンチパターンのおさらい • Python動的型付けの落とし穴 ▪ kwargs編 ▪ DataFrame編
55. Pydanticによる秩序の導入 • Pythonのデータ検証と シリアライゼーションの ためのライブラリ。 • 型ヒントを活用してラン タイムでのデータ検証を 行い、データの整合性を 保証。
56. Pydanticによる秩序の導入 • Pythonのデータ検証と シリアライゼーションの ためのライブラリ。 • 型ヒントを活用してラン タイムでのデータ検証を 行い、データの整合性を 保証。
57. before # extract.py def extract_data(user_id, **kwargs): start_date = kwargs.get("start_date", '2025-09-27') end_date = kwargs.get("end_date", datetime.now()) include_deleted = kwargs.get('include_deleted', False) batch_size = kwargs.get('batch_size', 1000) retry_count = kwargs.get('retry_count', 3) # さらに続く... 😱 どんなパラメータを渡せるの？ 😱 利用可能なオプション覚えておくの？
58. After from pydantic import BaseModel, Field from datetime import datetime class Param(BaseModel): user_id: int start_date: datetime = Field(default=datetime(2025, 9, 27)) end_date: datetime = Field(default_factory=datetime.now) include_deleted: bool = False batch_size: int = Field(default=1000, gt=0, le=10000) retry_count: int = Field(default=3, ge=1, le=5) def extract_data(param: Param): # 何が渡ってくる型が明確 😀 # IDEの補完が効く！　😀 # validな値が渡ってくる 😀
59. ipythonで動かしてみる
60. 正常想定の値によるインスタンス生成 : 生成成功 In [1]: from pydantic import BaseModel, Field ...: from datetime import datetime ...: class Param(BaseModel): ...: user_id: int ...: start_date: datetime = Field(default=datetime(2025, 9, 27)) ...: end_date: datetime = Field(default_factory=datetime.now) ...: include_deleted: bool = False ...: batch_size: int = Field(default=1000, gt=0, le=10000) ...: retry_count: int = Field(default=3, ge=1, le=5) In [2]: p = Param(user_id=123, start_date="2025-09-27T14:50:00") In [3]: p Out[3]: Param(user_id=123, start_date=datetime.datetime(2025, 9, 27, 14, 50), end_date=datetime.datetime(2025, 9, 14, 18, 10, 6, 756829), include_deleted=False, batch_size=1000, retry_count=3) 成功
61. 不正な値によるインスタンス生成 : 生成失敗 ---->1 param = Param( # 不正な値を渡すと 2 user_id="not_a_number" , # ❌ ValidationError 3 batch_size=100000) # ❌ batch_size must be <= 10000 File /usr/local/lib/python3.12/site-packages/pydantic/main.py:253, in BaseModel. __init__(self, **data) --> 253 validated_self = self.__pydantic_validator__.validate_python(data, self_instance =self) 254 if self is not validated_self: 255 warnings.warn( "省略", stacklevel=2) ValidationError : 2 validation errors for Param user_id Input should be a valid integer , unable to parse string as an integer [ type=int_parsing , input_value= 'not_a_number' , input_type= str] For further information visit https://errors.pydantic.dev /2.11/v/int_parsing batch_size "省略” 例外
62. IDEの補完
63. kwargs版
64. Pydantic版
65. ちなみに、イミュータブル(変更不可能)にできます
66. Pydantic デフォルト設定 In [1]: from pydantic import BaseModel, ConfigDict In [2]: class User(BaseModel): ...: id: int ...: name: str In [3]: u = User(id=123, name="PyCon2025") In [4]: u.name="hoge" In [5]: u Out[5]: User(id=123, name='hoge') 項目値変更可能 インスタンス生成成功
67. Pydantic イミュータブル設定 In [1]: from pydantic import BaseModel, ConfigDict ...: class ImmutableModel(BaseModel): ...: model_config = ConfigDict(frozen=True) # イミュータブル In [2]:class User(ImmutableModel): ...: id: int ...: name: str In [3]: u = User(id=123, name="PyCon2025") In [4]: u.name = "hoge" --------------------------------------------------------------------------- ValidationError Traceback (most recent call last) Cell In[4], line 1 ----> 1 u.name = "hoge" インスタンス生成成功 項目値変更すると例外
68. Python標準ライブラリのdataclass (型検証文脈での比較)
69. 型ヒントのみ In [1]: from dataclasses import dataclass ...: @dataclass ...: class Param: ...: name: str ...: age: int ...: In [2]: p = Param(name="PyCon2025", age="15歳") In [3]: p Out[3]: Param(name='PyCon2025', age='15歳') ”15歳”という文字列がそのまま格納 🤔
70. 他の選択肢はあるの?
71. 他の選択肢はあるの? import pandas as pd import pandera as pa # 「name」と「age」カラムが必須であるスキーマを定義 schema = pa.DataFrameSchema({ "name": pa.Column(str), # ageは0以上の整数 "age": pa.Column(int, pa.Check.ge(0)) }) invalid_df = pd.DataFrame({ "name": ["Taro", "Jiro"] # "age"カラムが欠けている！ }) try: schema.validate(invalid_df) except pa.errors.SchemaError as e: print(e)
72. Step by Stepアプローチ • Step1 : 型ヒントを追加してみる(割愛) • Step2 : Pydanticで1つの関数の引数を整理 • Step3 : レイヤー分離を試す • Step4 : チーム全体で改善の輪をひろげる
73. アンチパターンのおさらい • アーキテクチャ不在 • グローバル関数羅列 😱 認知負荷が高い 😱 変更の影響範囲がわかりにくい 😱 テストを書きにくい
74. アンチパターンのおさらい • アーキテクチャ不在 • グローバル関数羅列 😱 認知負荷が高い 😱 変更の影響範囲がわかりにくい 😱 テストを書きにくい
75. アンチパターンのおさらい • アーキテクチャ不在 • グローバル関数羅列 😱 認知負荷が高い 😱 変更の影響範囲がわかりにくい 😱 テストを書きにくい
76. アンチパターンのおさらい • アーキテクチャ不在 • グローバル関数羅列 😱 認知負荷が高い 😱 変更の影響範囲がわかりにくい 😱 テストを書きにくい
77. • リポジトリパターン • データアクセスロジックをビジネスロジックから分 離する設計パターン。 • データストア（データベース、API、ファイルシステ ムなど）へのアクセスを抽象化し、ドメインモデル とデータソースの間に仲介層を設けることで、アプ リケーションの保守性と拡張性を向上させます。 リポジトリパターンによる関心の分離
78. • リポジトリパターン • データアクセスロジックをビジネスロジックから分 離する設計パターン。 • データストア（データベース、API、ファイルシステ ムなど）へのアクセスを抽象化し、ドメインモデル とデータソースの間に仲介層を設けることで、アプ リケーションの保守性と拡張性を向上させます。 リポジトリパターンによる関心の分離
79. リポジトリパターンによる関心の分離 抽象定義 具象実装 ビジネスロジック
80. class IReportRepository(ABC): @abstractmethod def find_by_report_id(self, report_id: int) -> Report | None: pass class ReportRepository(IReportRepository): def __init__(self, session: Session): self._session = session # DB接続は外部から注入 def find_by_report_id(self, report_id: int) -> Report | None: x = self._session.query(ReportDTO).filter_by(id=report_id).one() #try except省略 return Report(report_id=x.id, report_name=x.report_name, expired_at=x.expired_at) class ReportUseCase: def __init__(self, repository: IReportRepository): self._repository = repository def find_report(self, report_id: int): report = self._repository.find_by_report_id(report_id) if not report: return None ... 省略 ... 抽象基底クラスを作成する仕組み 継承先にメソッド実装を強制
81. class IReportRepository(ABC): @abstractmethod def find_by_report_id(self, report_id: int) -> Report | None: pass class ReportRepository(IReportRepository): def __init__(self, session: Session): self._session = session # DB接続は外部から注入 def find_by_report_id(self, report_id: int) -> Report | None: x = self._session.query(ReportDTO).filter_by(id=report_id).one() #try except省略 return Report(report_id=x.id, report_name=x.report_name, expired_at=x.expired_at) class ReportUseCase: def __init__(self, repository: IReportRepository): self._repository = repository def find_report(self, report_id: int): report = self._repository.find_by_report_id(report_id) if not report: return None ... 省略 ... 継承元の@abstractmethod由来で実装を強制
82. before ## before: 混沌 query.py (1000行超え) ├── find_report_by_date_query() ├── find_report_by_date() ├── find_report_by_id_query() ├── find_report_by_id() └── ... (延々と続く)
83. after ## After: レイヤー構造 ├── cli # エントリーポイント ├── domain │ ├── entity │ └── repository # interfaceのみ ├── infra # DB接続など │ └── mysql │ ├── model │ └── repository # interfaceの実装
84. • 一貫性のあるデータアクセス • 関心の分離 • テストの容易性 リポジトリパターンの目的と利点
85. • 一貫性のあるデータアクセス • 関心の分離 • テストの容易性 リポジトリパターンの目的と利点
86. • 一貫性のあるデータアクセス • 関心の分離 • テストの容易性 リポジトリパターンの目的と利点
87. • 一貫性のあるデータアクセス • 関心の分離 • テストの容易性 リポジトリパターンの目的と利点
88. Step by Stepアプローチ • Step1 : 型ヒントを追加してみる(割愛) • Step2 : Pydanticで1つの関数の引数を整理 • Step3 : レイヤー分離を試す • Step4 : チーム全体で改善の輪をひろげる
89. チームの巻き込み方
90. • 1マイクロサービス -> 1人が担当 • コードレビューの文化なし 運用体制のおさらい
91. ①敬意から始める • 既存コードは「ビジネスに貢献してきた資産」 • 動いているコードへの感謝を忘れない • 批判ではなく「さらに良くする」スタンス
92. “あらゆる人間関係の衝突は、謙虚・尊敬・信頼の欠如によるものだ。”
93. • 改善ポイントを公開の場で議論 ◦ Slack/Teamsのpublic channel or GitHub ◦ 定例でのアジェンダ化 ◦ 段階的な合意形成 ▪ Step1 : 課題の共有 ▪ Step2 : 解決策の提案 ▪ Step3 : 小さく試す ②透明性のある進め方
94. ③チーム全体の成長機会に • アーキテクチャ共有会の開催 ◦ Before/Afterの比較 ◦ 改善の意図を説明 ◦ 質問・議論の時間を十分に ◦ ペアプロ・モブプロを取り入れるなど
95. 小さく始めて大きく育てる
96. Step by Stepアプローチ(おさらい） • Step1 : 型ヒントを追加してみる • Step2 : Pydanticで1つの関数の引数を整理 • Step3 : レイヤー分離を試す • Step4 : チーム全体で改善の輪をひろげる
97. まとめ
98. ご清聴ありがとうございました 技術は「きっかけ」、文化が「ゴール」 アーキテクチャを育てるのはチームです。