Python スクリプト運用を効率化するための手引き その２
はじめに
前に書いた奴が思いのほか評価されているので調子に乗って書きます、というのが本心です。建前は「前作は些か一般論が過ぎていないか？このぐらいの話なんて自明じゃないか？」と思ったのでもう少し色々書こう、といった具合です。
前提
- ヘビ (Python) 使いな人。ぴーぴょろ笛を吹いてヘビを好き勝手に使役できると尚良いです。
- Python 3.13 で考えるようにしてください
- 前作で触れたように uv を使うのでそのつもりでお願いします
- 別に環境差はないと思うけど、私のお手元検証環境は Ubuntu 24.04 です
誰向け？
- Python にいくらか詳しい人。あんまり詳しくない人は詳しくなってください
- ただでさえ忙しいのに 業務効率化スクリプトの実装や待ち時間に時間を費やしたくない人
- つまり、業務効率化を業務効率化してやろうというのがこの記事の魂胆です
- 前作を読んでくれてると嬉しいですが、読んでいなくても独立した話なので読んでなくても嬉しいです
キモ（スクリプト実装編）
クレデンシャルを安全に扱う
絶対に症状でググるな並に大事にしてほしいのがクレデンシャルの取り扱いです。下手こいてコミットしたりプッシュしたりすると大変なことになるので、予防措置をするに越したことはないです。ただ、例えば.gitignore に .env とかを入れておけってのは死ぬほど言われていると思うので割愛します。問題はどう読み込むかだよ。Secrets Manager とか Vault とかを使えれば良いんですけど、それこそ仰々しくない？
ともかく。特に環境変数もとい .env をぱいてょんで取り扱う方法は大きく２つあります。片割れの一人は python-dotenv、もう一人の片割れは pydantic-settings です。前者は中々に雑っぽい環境でも雑にセットアップしてオッケー、後者は Pydantic の堅牢なモデリングに環境変数を導入できて後々楽なのが利点です。
作業タスクごとにスクリプトは分割する
例えば「スプレッドシートをダウンロードする → pl.read_excel() でどうにかする → どうにかした DataFrame を df.write_csv() する」というシナリオを想定します。 このとき、
- ダウンロードロジックはそんな実装に時間を食わない(使い回せるため)割に実行時間はかかる
- スプレッドシートをパースしていい感じに加工するのは実装が手間だが、実行はだいたいマジで秒
みたいな図式が成り立つと思います。なんで後者に実装時は工数を費やすことになるので、後者だけアジャイルよろしくろくろ実行を回せるようにスクリプトを組んどきましょう。 別にスクリプトを例示するほどじゃないですが、上記の例なら uv run download.py と uv run parse_and_export.py みたいにしたら良いと思います。
アーティファクトは全部ファイルとして保存する
どうでもいいけど成果物って言い方ちょっとあれじゃないですか？成果じゃないのに出来上がっただけで成果物って。なのでここでは、スクリプトの実行途上にできたあれこれをアーティファクトって言うことにします。
で、どうせあんたは調査やデバッグをするときに何がどうなってんのか調べることになります。だいたい何らかのファイルとにらめっこすることになるので、にらめっこしやすくするために用なアーティファクトはファイルとして保存しときましょう。
前回触れたかもわからん pathlib.Path はこういうときマジで便利です。 場合によっては shutil（高水準なファイルパスのコピーや移動を司る標準ライブラリ） とか組み合わせても良いかも。
from pathlib import Path
from loguru import logger
# こうすれば uv run の実行ディレクトリに関わらず常に一定のパスが取れます
# Jupyter じゃ動かないんでそこだけ注意
root_dir = Path(__file__).parent
# `/` を使ってファイルパスを結合できます（存在は保証されません）
text_path = (root_dir / "test.txt")
# 読み書きもできます 書く分にはいいけど読むのはファイルがないと例外を吐いて死ぬ
# ここでは問題ないが、root_dir が存在しないと write_text も例外を吐いて死ぬ
text_path.write_text("Hello, World!")
logger.info(text_path.read_text())
また、現代のコンピューを実践しているモダンなライブラリ、たとえば polars (pandas のハイテクエディション) は pathlib.Path をそのまま受け取れます。データフレームも立派なアーティファクトなので保存しときましょう。
pl.DataFrame([
{"name": "Alice", "age": 30},
{"name": "Bob", "age": 25},
]).write_csv(root_dir / "test.csv")
ただ注意点。pathlib はブロッキングIOな奴なので非同期処理と組み合わせるとちょっと相性が悪かったりします。もしそのような境地にたどり着いたら、anyio や aiofiles が頼りになるはずです。 aiofiles は若干機能がたりないフシがあって、翻って anyio のほうはなにかと使い勝手が良い (非同期化された pathlib.Path 相当の anyio.Path があり、いくらか高機能な TaskGroup もある等) のでこれから使うなら anyio がいいかも。
引数で実行時のパラメータを指定できるようにする
これはスクリプトの実行時に引数を受け取って好き放題したいなら ArgumentParser を使いやがれってことです。 optparse だか getopt だかのほうが良いとする向きもあるんですが、別に ArgumentParser で困ることはなく、実際ほぼデファクトスタンダードなのでこれで良いと思います。
from argparse import ArgumentParser
def main() -> None:
# description とか help は -h オプションを呼び出すと出てきます
parser = ArgumentParser(description="A GREATEST PROGRAM EVER!")
parser.add_argument("-n", "--name", type=str, default="ARGIA", help="The name to greet.")
args = parser.parse_args()
greet(args.name)
def greet(name: str) -> None:
print(f"Hello, {name}!")
if __name__ == "__main__":
main()
--help かショートハンドの -h を使うとヘルプを出せます。実行例をまとめると以下のとおりです。
❯ uv run arg.py -h
usage: arg.py [-h] [-n NAME]
A GREATEST PROGRAM EVER!
options:
-h, --help show this help message and exit
-n, --name NAME The name to greet.
❯ uv run arg.py -n AOI
Hello, AOI!
また、parser.add_argument はフラグも扱えます。例えば dry-run 的なロジックを組み込みたいとして、dry-run な状態を受け取りたいときは以下のようにすればいいです。このとき、-d を指定しなかった場合は args.dry_run は False になります。
parser.add_argument("-d", "--dry-run", action="store_true", help="...")
非同期処理化できる処理は、する
正確性はちょっとあの世に送って、雑に話をします。
ぱいてょんは async def でコルーチン的なものを作れます。 で、async できるということは await できるということです。 API コールはたいてい外部の待ち時間が多いので、async def を使っていい感じに非同期処理を書いてやって、できあがったコルーチン御一行様を並行で捌いてやれば結構な時短になります。
前作のお作法に則って書くと、以下のようになります。 現代的も方法を使うといいです。たとえば asyncio.gather とかを使うのがセオリーっぽいですが、今風なのは asyncio.TaskGroup です。並列処理は勝手にされるのでそこは考えなくてオッケー。
細かい話は技術評論社の記事が詳しいです（例外処理とかコルーチンのキャンセル処理が楽になります）。
import asyncio
from loguru import logger
async def main() -> None:
try:
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(do_coroutine(i)) for i in range(10)]
results = [task.result() for task in tasks]
# エラーが起きなきゃ result は list[int] です
logger.info(results)
# tasks 内のコルーチンで例外が吐かれたら、例外をグループ化して拾えます
except* Exception as eg:
for e in eg.exceptions:
logger.error(f"An error occurred: {e}")
async def do_coroutine(i: int) -> int:
await asyncio.sleep(3)
if i in (3, 6):
msg = f"An error occurred in do_coroutine: {i=}"
raise ValueError(msg)
return i
if __name__ == "__main__":
asyncio.run(main())
でも、これはなんか仰々しいし、みんな大好き tqdm との連携はどうするの？ しかも loguru と tqdm を組み合わせて実行すると見た目やべーけど？って話もあるので、その場合は以下のようにできます。例外処理はちょっとがんばってください。
import asyncio
import random
import tqdm
from loguru import logger
# デフォルトのログ出力を犠牲に、tqdm 経由でログを吐き出すようにする
# ターミナルで見る分には特に問題は生まれないはず
logger.remove()
logger.add(lambda s: tqdm.tqdm.write(s, end=""), colorize=True)
async def main() -> None:
tasks = [do_coroutine(i) for i in range(10)]
for task in tqdm.tqdm(asyncio.as_completed(tasks), total=len(tasks), desc="Processing"):
await task
async def do_coroutine(i: int) -> int:
delay = random.uniform(1, 3)
await asyncio.sleep(delay)
logger.info(f"Finished do_coroutine: {i=}, delay={delay:.2f}s")
return i
if __name__ == "__main__":
asyncio.run(main())
実行結果は 🦑 のような感じです。
❯ uv run korede-iiyo.py
2026-02-21 00:25:02.556 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=9, delay=1.07s
2026-02-21 00:25:02.677 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=7, delay=1.19s
2026-02-21 00:25:02.915 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=4, delay=1.43s
2026-02-21 00:25:03.325 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=1, delay=1.84s
2026-02-21 00:25:03.388 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=3, delay=1.90s
2026-02-21 00:25:03.564 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=8, delay=2.08s
2026-02-21 00:25:04.013 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=0, delay=2.53s
2026-02-21 00:25:04.210 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=6, delay=2.73s
2026-02-21 00:25:04.340 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=2, delay=2.86s
2026-02-21 00:25:04.443 | INFO | __main__:do_coroutine:22 - Finished do_coroutine: i=5, delay=2.96s
Processing: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 10/10 [00:02<00:00, 3.38it/s]
ちなみに上記スクリプトを見ると「async def に渡す引数と結果をペアにして for するにはどうしたら？」とか「コルーチンの実行結果順を保証したいんだけどどうしたら？」（※ asyncio.as_completed は gather みたいに実行結果順が保証されない) みたいな話が出てくると思います。前者は async def のリターンを tuple にするとかしてください。後者は……チャッピーに聞いてください（あんまりこのユースケースに出会ったためしがないため）。あと Python 3.14 は忌々しき GIL から解放されてヒャッハー！みたいなこともあるらしいんですが、これも右に同じでジェミニに聞いてください。ここではカバーしません。
非同期処理をするときは asyncio.Semaphore する
さもなくば API コールをしまくったときにレートリミットやねんと言い放たれて HTTP 429 と共に絶命したり、そもそもスクリプト側でコルーチンを積みすぎてまともに処理が進まなかったりとかそういう弊害が生まれることでしょう。以下はセマフォで最大並列=3に制限してみた例です。実際の数字は API の同時コール制限数や実際の負荷などを見て調整すると良いと思います。経験則的には良くて10が最大です。
import asyncio
async def main() -> None:
async with asyncio.TaskGroup() as tg:
semaphore = asyncio.Semaphore(3)
for i in range(12):
tg.create_task(do_coroutine(i, semaphore))
async def do_coroutine(i: int, semaphore: asyncio.Semaphore) -> int:
async with semaphore:
await asyncio.sleep(3)
return i
if __name__ == "__main__":
asyncio.run(main())
実行すると以下の通りになります。上記のスクリプトを読み解くと、コルーチンの実行はセマフォで最大3並列に制限されており、コルーチンは12個実行され、1個あたり3秒かかるので、つまり全体の実行時間はだいたい12秒になるはずです。なりました。
❯ time uv run sem.py
real 0m12.119s
user 0m0.062s
sys 0m0.043s
おわりに
最近の作業用 BGM のヘビロテ枠はテクノトリスです。 テクノ名称枠に限って言うと時点はテクノポリスですが、こっちはそこそこ好きってぐらいです。似たような語感枠で言うとメトロポリスも好きです。 ひどいスクリプトを書いているときのテンション盛り上げ（盛り下げ）ソングにいかがでしょうか。
Discussion