Python pathlibを使った業務ファイル処理の実践パターン集
Pythonでファイル操作を行う際、かつてはosモジュールが標準でしたが、現在はpathlibの利用が推奨されています。実務では、ファイルの読み書き、ディレクトリ構造の管理、設定ファイルの処理など、様々なシーンでパス操作が必要になります。本記事では、業務で実際に使えるpathlibのパターンを詳しく解説します。
pathlibの簡易的な解説
pathlibはPython 3.4で導入されたモジュールで、オブジェクト指向的にパスを扱えます。従来のos.path.joinやos.path.isdir()といった関数型のアプローチではなく、Pathオブジェクトのメソッドを使用します。
基本的な使い方は以下の通りです：
from pathlib import Path
# パスオブジェクトの作成
file_path = Path('data/input.csv')
parent_dir = file_path.parent
file_name = file_path.name
# パスの結合（/演算子を使用）
new_path = Path('data') / 'output' / 'result.csv'
# ファイルの存在確認
if file_path.exists():
print(f'{file_path}は存在します')
# ファイルかディレクトリか判定
if file_path.is_file():
print('ファイルです')
elif file_path.is_dir():
print('ディレクトリです')
pathlibの大きな利点は、OSに依存しないパス操作ができることです。Windowsではバックスラッシュ、LinuxやMacではスラッシュといった違いを自動的に処理してくれます。
業務でのユースケース
ユースケース1：ログファイルの日次処理
実務では、毎日生成されるログファイルを処理することがあります。特定のディレクトリから本日のログファイルを見つけ出し、処理結果を別のディレクトリに保存するという典型的なパターンです。
ユースケース2：設定ファイルの検索と読み込み
アプリケーション起動時に、複数の可能性のあるディレクトリから設定ファイルを探す必要があります。ホームディレクトリ、カレントディレクトリ、システムディレクトリなどを順番に確認するパターンです。
ユースケース3：バッチ処理でのファイル一括変換
大量のCSVファイルやJSONファイルを処理して、形式を変換したり、データを抽出したりするバッチ処理では、glob()を使ったパターンマッチングが活躍します。
実装コード：実務パターン集
パターン1：ログファイルの日次処理
実務では、このようなログ処理はスケジュール実行されることが多いです。
from pathlib import Path
from datetime import datetime, timedelta
import logging
class LogProcessor:
def __init__(self, log_dir: str, archive_dir: str):
self.log_dir = Path(log_dir)
self.archive_dir = Path(archive_dir)
self.archive_dir.mkdir(parents=True, exist_ok=True)
def find_logs_by_date(self, target_date: datetime = None):
"""指定日のログファイルを検索"""
if target_date is None:
target_date = datetime.now()
date_str = target_date.strftime('%Y%m%d')
log_pattern = f'app_{date_str}_*.log'
matching_logs = list(self.log_dir.glob(log_pattern))
return matching_logs
def process_daily_logs(self, target_date: datetime = None):
"""日次ログを処理してアーカイブ"""
logs = self.find_logs_by_date(target_date)
if not logs:
print(f'{target_date.date()}のログファイルがありません')
return
# ログファイルを処理
total_errors = 0
total_warnings = 0
for log_file in logs:
with open(log_file, 'r', encoding='utf-8') as f:
for line in f:
if 'ERROR' in line:
total_errors += 1
elif 'WARNING' in line:
total_warnings += 1
# 結果レポートを生成
report_path = self.archive_dir / f'report_{target_date.strftime("%Y%m%d")}.txt'
with open(report_path, 'w', encoding='utf-8') as f:
f.write(f'Log Report for {target_date.date()}\n')
f.write(f'Total Errors: {total_errors}\n')
f.write(f'Total Warnings: {total_warnings}\n')
# ログファイルをアーカイブディレクトリに移動
archive_subdir = self.archive_dir / target_date.strftime('%Y/%m')
archive_subdir.mkdir(parents=True, exist_ok=True)
for log_file in logs:
log_file.rename(archive_subdir / log_file.name)
print(f'処理完了: {len(logs)}個のファイルをアーカイブしました')
# 使用例
processor = LogProcessor('logs', 'archives')
processor.process_daily_logs()
パターン2：設定ファイルの検索と読み込み
実際のアプリケーションでは、複数の設定ファイルの候補場所から最初に見つかったものを使用します。
from pathlib import Path
import json
from typing import Optional, Dict, Any
class ConfigManager:
def __init__(self, config_name: str = 'config.json'):
self.config_name = config_name
self.config_paths = [
Path.home() / '.config' / 'myapp' / config_name,
Path.cwd() / config_name,
Path.cwd() / 'config' / config_name,
Path('/etc/myapp') / config_name,
]
def find_config(self) -> Optional[Path]:
"""最初に見つかった設定ファイルを返す"""
for config_path in self.config_paths:
if config_path.exists() and config_path.is_file():
print(f'設定ファイルを発見: {config_path}')
return config_path
return None
def load_config(self) -> Dict[str, Any]:
"""設定ファイルを読み込む"""
config_file = self.find_config()
if config_file is None:
print('警告：設定ファイルが見つかりません。デフォルト値を使用します')
return self._default_config()
try:
with open(config_file, 'r', encoding='utf-8') as f:
config = json.load(f)
return config
except json.JSONDecodeError as e:
print(f'エラー：設定ファイルのパースに失敗しました: {e}')
return self._default_config()
def _default_config(self) -> Dict[str, Any]:
"""デフォルト設定を返す"""
return {
'debug': False,
'log_level': 'INFO',
'database': 'sqlite:///app.db',
}
# 使用例
config_manager = ConfigManager()
config = config_manager.load_config()
print(f'ログレベル: {config["log_level"]}')
パターン3：バッチファイル変換処理
大量のファイルを処理する際は、glob()でファイルパターンマッチングを行い、効率的に処理します。
from pathlib import Path
import csv
import json
from typing import List
class BatchFileConverter:
def __init__(self, input_dir: str, output_dir: str):
self.input_dir = Path(input_dir)
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
def convert_csv_to_json(self) -> List[str]:
"""入力ディレクトリ内のすべてのCSVをJSONに変換"""
converted_files = []
# 再帰的にCSVファイルを検索
csv_files = list(self.input_dir.rglob('*.csv'))
print(f'{len(csv_files)}個のCSVファイルが見つかりました')
for csv_file in csv_files:
try:
# 相対パスを保持してディレクトリ構造を再現
relative_path = csv_file.relative_to(self.input_dir)
json_path = self.output_dir / relative_path.with_suffix('.json')
json_path.parent.mkdir(parents=True, exist_ok=True)
# CSV読み込み
rows = []
with open(csv_file, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
rows = list(reader)
# JSON書き込み
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(rows, f, ensure_ascii=False, indent=2)
converted_files.append(str(json_path))
print(f'✓ 変換完了: {relative_path}')
except Exception as e:
print(f'✗ エラー（{csv_file}）: {e}')
return converted_files
def get_file_stats(self) -> dict:
"""入力ディレクトリ内のファイル統計を取得"""
stats = {
'csv_files': len(list(self.input_dir.rglob('*.csv'))),
'json_files': len(list(self.input_dir.rglob('*.json'))),
'txt_files': len(list(self.input_dir.rglob('*.txt'))),
'total_size_mb': 0,
}
# 全ファイルのサイズ合計を計算
total_size = sum(f.stat().st_size for f in self.input_dir.rglob('*') if f.is_file())
stats['total_size_mb'] = round(total_size / (1024 * 1024), 2)
return stats
# 使用例
converter = BatchFileConverter('input_data', 'output_data')
converter.convert_csv_to_json()
stats = converter.get_file_stats()
print(f'統計: {stats}')
パターン4：テンポラリファイルの安全な管理
業務では一時的なファイルを安全に処理する必要があります。pathlibと併用できるパターンです。
from pathlib import Path
import tempfile
from contextlib import contextmanager
class TemporaryFileManager:
@staticmethod
@contextmanager
def temporary_directory(base_dir: str = None):
"""一時ディレクトリを管理するコンテキストマネージャ"""
if base_dir:
temp_dir = Path(base_dir) / 'temp'
temp_dir.mkdir(parents=True, exist_ok=True)
temp_path = Path(tempfile.mkdtemp(dir=temp_dir))
else:
temp_path = Path(tempfile.mkdtemp())
try:
yield temp_path
finally:
# クリーンアップ
import shutil
shutil.rmtree(temp_path)
@staticmethod
def process_with_backup(source_file: str, processor_func):
"""ファイル処理中にバックアップを作成"""
source_path = Path(source_file)
if not source_path.exists():
raise FileNotFoundError(f'{source_path}が見つかりません')
# バックアップ作成
backup_path = source_path.with_suffix(source_path.suffix + '.bak')
import shutil
shutil.copy2(source_path, backup_path)
try:
# 処理実行
result = processor_func(source_path)
print(f'処理成功: {source_path}')
# バックアップ削除
backup_path.unlink()
return result
except Exception as e:
print(f'処理失敗。バックアップから復元します')
shutil.copy2(backup_path, source_path)
backup_path.unlink()
raise e
# 使用例
with TemporaryFileManager.temporary_directory('work') as temp_dir:
temp_file = temp_dir / 'temp_data.txt'
temp_file.write_text('一時データ')
print(f'テンポラリファイル作成: {temp_file}')
# ここでの処理後、自動的にクリーンアップされる
よくある応用パターン
パターン1：ファイル監視と自動処理
特定のディレクトリにファイルが追加されたら自動的に処理するパターンです。業務ではこのような要件が頻繁に出ます。
from pathlib import Path
import time
from datetime import datetime
class FileWatcher:
def __init__(self, watch_dir: str, process_func):
self.watch_dir = Path(watch_dir)
self.process_func = process_func
self.processed_files = set()
def watch(self, interval: int = 5, max_iterations: int = None):
"""ディレクトリを監視して新しいファイルを処理"""
iteration = 0
while True:
try:
current_files = set(self.watch_dir.glob('*.*'))
new_files = current_files - self.processed_files
for new_file in new_files:
if new_file.is_file():
print(f'[{datetime.now()}] 新規ファイル検出: {new_file.name}')
self.process_func(new_file)
self.processed_files.add(new_file)
iteration += 1
if max_iterations and iteration >= max_iterations:
break
time.sleep(interval)
except KeyboardInterrupt:
print('\n監視を終了します')
break
# 使用例
def process_incoming_file(file_path: Path):
print(f'処理中: {file_path}')
# ファイルの内容を処理
watcher = FileWatcher('incoming', process_incoming_file)
# watcher.watch()
パターン2：ディレクトリのバックアップと復元
本番環境に関わるシステムでは、変更前のバックアップが重要です。
from pathlib import Path
import shutil
from datetime import datetime
class BackupManager:
def __init__(self, backup_base_dir: str):
self.backup_base_dir = Path(backup_base_dir)
self.backup_base_dir.mkdir(parents=True, exist_ok=True)
def create_backup(self, target_dir: str, backup_name: str = None) -> Path:
"""ディレクトリのバックアップを作成"""
target_path = Path(target_dir)
if not target_path.exists():
raise FileNotFoundError(f'{target_path}が見つかりません')
if backup_name is None:
backup_name = f'{target_path.name}_{datetime.now().strftime("%Y%m%d_%H%M%S")}'
backup_path = self.backup_base_dir / backup_name
if target_path.is_dir():
shutil.copytree(target_path, backup_path)
else:
shutil.copy2(target_path, backup_path)
print(f'バックアップ作成: {backup_path}')
return backup_path
def list_backups(self, filter_name: str = None) -> list:
"""利用可能なバックアップの一覧を表示"""
backups = list(self.backup_base_dir.iterdir())
if filter_name:
backups = [b for b in backups if filter_name in b.name]
backups.sort(key=lambda x: x.stat().st_mtime, reverse=True)
return backups
def restore_backup(self, backup_name: str, restore_to: str):
"""バックアップを復元"""
backup_path = self.backup_base_dir / backup_name
restore_path = Path(restore_to)
if not backup_path.exists():
raise FileNotFoundError(f'バックアップ{backup_name}が見つかりません')
# 既存ファイルをバックアップ
if restore_path.exists():
temp_backup = self.backup_base_dir / f'{restore_path.name}_restore_backup'
if restore_path.is_dir():
shutil.copytree(restore_path, temp_backup)
else:
shutil.copy2(restore_path, temp_backup)
# 復元
if backup_path.is_dir():
if restore_path.exists():
shutil.rmtree(restore_path)
shutil.copytree(backup_path, restore_path)
else:
restore_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(backup_path, restore_path)
print(f'復元完了: {backup_name} -> {restore_to}')
# 使用例
backup_mgr = BackupManager('backups')
backup_mgr.create_backup('data')
for backup in backup_mgr.list_backups():
print(f'- {backup.name}')
注意点
1. エンコーディングの明示的指定
テキストファイルを読み書きする際は、エンコーディングを明示的に指定しましょう。特に業務システムでは、異なるOS環境での動作を想定する必要があります。
from pathlib import Path
# 良い例
with open(Path('data.txt'), 'r', encoding='utf-8') as f:
content = f.read()
# 避けるべき例
with open(Path('data.txt'), 'r') as f:
content = f.read() # OSのデフォルトエンコーディングに依存
2. パス操作時の例外処理
ファイル操作は実行環境によって失敗する可能性があります。適切な例外処理が必須です。
from pathlib import Path
try:
file_path = Path('important_file.txt')
content = file_path.read_text(encoding='utf-8')
except FileNotFoundError:
print('ファイルが見つかりません')
except PermissionError:
print('ファイルにアクセスする権限がありません')
except Exception as e:
print(f'予期しないエラー: {e}')
3. resolve()の活用
相対パスを絶対パスに変換するには、resolve()を使用します。これにより、シンボリックリンクも解決されます。
from pathlib import Path
# 相対パスの場合
relative_path = Path('data/file.txt')
absolute_path = relative_path.resolve()
print(absolute_path)
# 例: /home/user/project/data/file.txt
4. exists()のTOCTOU問題
exists()でファイル存在確認後に、ファイルを削除されるという競合状態が発生する可能性があります。実務では以下のように対応します。
from pathlib import Path
file_path = Path('important_file.txt')
try:
# 存在確認ではなく、直接操作して例外処理
content = file_path.read_text(encoding='utf-8')
except FileNotFoundError:
print('ファイルが見つかりません')
except Exception as e:
print(f'エラー: {e}')
5. glob()のパフォーマンス
大量のファイルがあるディレクトリでglob()を使う場合、パフォーマンスに注意が必要です。必要に応じてフィルタリングを工夫しましょう。
from pathlib import Path
dir_path = Path('large_directory')
# 非効率：すべてのファイルを取得してからフィルタリング
all_files = dir_path.glob('*')
large_files = [f for f in all_files if f.stat().st_size > 1000000]
# より効率的：glob後にフィルタリング
large_files = [f for f in dir_path.glob('*.log') if f.stat().st_size > 1000000]
まとめ
pathlibはPythonの標準ライブラリとして、業務システム開発において非常に重要な役割を果たします。本記事で紹介した5つの実装パターン（ログ処理、設定管理、バッチ変換、テンポラリファイル管理、ファイル監視）は、実際のプロジェクトで頻繁に出現する要件です。
pathlibを使うことで、以下のメリットが得られます：
- 可読性の向上：オブジェクト指向的なAPI設計により、意図が明確になります
- クロスプラットフォーム対応：OSの違いを意識する必要がありません
- 保守性の向上：パス操作がシンプルに記述でき、バグが減少します
- 開発効率の向上：豊富なメソッドにより、複雑なパス操作も直感的に実装できます
実務でファイル操作を行う際は、osモジュールではなくpathlibを優先的に選択することをお勧めします。特に新規プロジェクトでは、最初からpathlibで統一すると、後々の保守がずっと楽になります。
また、エラーハンドリングと例外処理を適切に実装することで、本番環境でのトラブルを未然に防ぐことができます。紹介したコード例を参考に、皆さんのプロジェクトに合わせてカスタマイズし、効率的なファイル処理システムを構築してください。
