Python例外処理(try-except)
2025-10-31はじめに
プログラミングにおいて、エラーは避けて通れない存在です。実際のプログラミングでは、構文エラーだけでなく、プログラム実行中に発生するさまざまな「実行時エラー」に対処する必要があります。
例外処理は、こうした実行時エラーを効果的に管理し、プログラムの堅牢性(robustness)を高めるための重要な技術です。本章では、Pythonの例外処理メカニズムであるtry-except構文を詳細に解説し、より信頼性の高いプログラムを作成する方法を学びます。
例外処理の基本概念
例外とは何か
例外(Exception)とは、プログラムの正常な実行フローを妨げるイベントのことを指します。例えば、存在しないファイルを開こうとしたり、ゼロで除算を行ったり、不正な型のデータを操作しようとした場合に発生します。

Pythonでは、例外が発生すると、通常のプログラムの流れが中断され、例外オブジェクトが「送出(raise)」されます。この例外が捕捉(catch)されない場合、プログラムはエラーメッセージを表示して終了します。
例外処理の重要性
例外処理を適切に行うことには、以下のような利点があります。
- プログラムのクラッシュ防止:予期しないエラーが発生してもプログラムを継続実行可能
- ユーザーフレンドリーなエラーメッセージの提供:技術的な詳細ではなく、理解しやすいメッセージを表示
- リソースの適切な管理:ファイルやネットワーク接続などのリソースを確実に解放
- デバッグの容易化:エラー発生時の状況を詳細に記録
try-except構文の基本
基本的な構文
Pythonにおける最も基本的な例外処理の構文は以下の通りです。tryブロック内でエラーが発生する可能性のある「危険な処理」を実行し、もし指定したExceptionTypeの例外が発生した場合はexceptブロックでそれを捕捉して「エラー処理」を行います。これにより、プログラムが予期せぬエラーで強制終了するのを防ぎ、安全に制御された処理を続けることができます。
try:
# 例外が発生する可能性のあるコード
危険な処理()
except ExceptionType:
# 例外が発生した場合の処理
エラー処理()
具体例:ゼロ除算の処理
実際の例を見てみましょう。ゼロ除算は典型的な実行時エラーです。divide_without_handling()はゼロ除算を行うとZeroDivisionErrorが発生し、プログラムが強制終了します。一方、divide_with_handling()ではtry-except構文を使って同じ処理を安全に実行し、ゼロで割ろうとした場合に「ゼロで除算することはできません」と表示してNoneを返します。これにより、エラー発生時でもプログラムを継続できます。
# 例外処理なしの場合
def divide_without_handling(a, b):
return a / b
# この呼び出しはZeroDivisionErrorを発生させる
# result = divide_without_handling(10, 0) # プログラムがクラッシュ
# 例外処理ありの場合
def divide_with_handling(a, b):
try:
result = a / b
return result
except ZeroDivisionError:
print("エラー: ゼロで除算することはできません")
return None
# 安全に実行可能
result = divide_with_handling(10, 0) # エラーメッセージを表示し、Noneを返す
複数の例外タイプの処理
実際のプログラムでは、異なる種類の例外が発生する可能性があります。それぞれの例外タイプに応じた処理を記述できます。tryブロックでa / bを実行し、除数が0の場合はZeroDivisionErrorを捕捉して「除数がゼロです」と表示します。数値以外の型が渡された場合はTypeErrorを捕捉し、「数値型以外の値が渡されました」と警告します。どちらのエラーでもNoneを返すため、プログラムが中断せずに安全に動作を続けられます。
def safe_divide(a, b):
try:
result = a / b
return result
except ZeroDivisionError:
print("エラー: 除数がゼロです")
return None
except TypeError:
print("エラー: 数値型以外の値が渡されました")
return None
# 様々なエラーケースのテスト
print(safe_divide(10, 2)) # 正常: 5.0
print(safe_divide(10, 0)) # ZeroDivisionErrorを捕捉
print(safe_divide(10, "2")) # TypeErrorを捕捉
一つのexcept節での複数例外の捕捉
関連する例外をまとめて処理することも可能です。tryブロックでfloat()変換と除算を行い、変換できない場合はValueErrorやTypeErrorを捕捉してエラーメッセージを表示します。入力が0のときはZeroDivisionErrorを捕捉し、「ゼロの逆数は定義できません」と警告します。どの例外も処理後にNoneを返し、プログラムが中断せず安定して動作します。
def process_number(value):
try:
number = float(value)
reciprocal = 1 / number
return reciprocal
except (ValueError, TypeError):
print("エラー: 数値に変換できない値が渡されました")
return None
except ZeroDivisionError:
print("エラー: ゼロの逆数は定義できません")
return None
# テスト
print(process_number(5)) # 正常: 0.2
print(process_number(0)) # ZeroDivisionError
print(process_number("abc")) # ValueError
例外オブジェクトの情報取得
例外発生時には、例外オブジェクトから詳細な情報を取得できます。tryブロックでa / bを実行し、bが0ならZeroDivisionErrorを捕捉してエラーメッセージと例外の型名を表示します。bが数値でない場合はTypeErrorを捕捉し、型の不正を報告します。いずれの場合もNoneを返し、安全にプログラムを継続できるようにしています。エラーの内容を明確に把握できる実践的な例です。
def detailed_error_handling(a, b):
try:
result = a / b
return result
except ZeroDivisionError as e:
print(f"ゼロ除算エラーが発生: {e}")
print(f"エラーの型: {type(e).__name__}")
return None
except TypeError as e:
print(f"型エラーが発生: {e}")
return None
# エラー情報を詳細に表示
detailed_error_handling(10, 0)
detailed_error_handling(10, "2")
else節の使用
tryブロックで例外が発生しなかった場合に実行されるコードをelse節に記述できます。tryブロックでファイルを開き、存在しない場合はFileNotFoundError、読み込み中に問題が発生した場合はIOErrorを捕捉してエラーメッセージを表示します。例外が起きなかった場合のみelseブロックが実行され、ファイルの内容を読み取って閉じた後、その内容を返します。エラー時も安全に処理を終了できる堅牢な設計です。
def read_file_safely(filename):
try:
file = open(filename, 'r')
except FileNotFoundError:
print(f"エラー: ファイル '{filename}' が見つかりません")
return None
except IOError as e:
print(f"エラー: ファイル読み込み中に問題が発生: {e}")
return None
else:
# 例外が発生しなかった場合のみ実行
content = file.read()
file.close()
return content
finally節の使用
例外の有無に関わらず必ず実行されるコードをfinally節に記述します。リソースの解放などに利用されます。tryブロックでファイルを開き内容を読み込みますが、ファイルが存在しない場合はFileNotFoundErrorで補足し、エラーメッセージを表示してNoneを返します。finallyブロックは例外の有無に関係なく実行され、開いたファイルを確実に閉じてリソースリークを防ぎます。安全で信頼性の高いファイル操作方法を示す典型例です。
def process_file(filename):
file = None
try:
file = open(filename, 'r')
content = file.read()
# 何らかの処理
return content
except FileNotFoundError:
print(f"ファイル '{filename}' が見つかりません")
return None
finally:
# 例外の有無に関わらず実行
if file is not None:
file.close()
print("ファイルを閉じました")
完全なtry-except-else-finally構造
これらを組み合わせた完全な構造は以下の通りです。tryでファイルを開いて内容を読み込み、exceptでFileNotFoundErrorやIOErrorを捕捉して適切なエラーメッセージを出力します。エラーがなければelseブロックで成功メッセージと内容を返し、最後にfinallyでファイルを閉じてリソースを確実に解放します。エラー発生時でも後処理を行う、安全で堅牢な例外処理の構造です。
def comprehensive_example(filename):
file = None
try:
file = open(filename, 'r')
content = file.read()
except FileNotFoundError:
print(f"エラー: ファイル '{filename}' が見つかりません")
return None
except IOError as e:
print(f"エラー: 入出力エラーが発生: {e}")
return None
else:
print("ファイルの読み込みに成功しました")
return content
finally:
if file is not None:
file.close()
print("リソースのクリーンアップを実行しました")
例外の階層と捕捉の順序
Pythonの例外は階層構造を持っています。このため、例外を捕捉する順序が重要になります。exception_hierarchy_example()関数では、文字列を整数に変換し、100を割り算しています。変換に失敗すればValueError、0で割ればZeroDivisionError(親クラスArithmeticErrorで捕捉)として処理されます。さらに、どの例外にも当てはまらない場合は、最上位のExceptionで処理されます。これにより、例外を階層的に扱う方法が理解できます。
def exception_hierarchy_example(value):
try:
number = int(value)
result = 100 / number
return result
except ValueError:
print("値の変換に失敗しました")
except ArithmeticError: # ZeroDivisionErrorの親クラス
print("算術エラーが発生しました")
except Exception as e: # すべての例外の親クラス
print(f"予期しないエラー: {e}")
# テスト
exception_hierarchy_example("abc") # ValueError
exception_hierarchy_example(0) # ArithmeticError (ZeroDivisionError)
実践的な例外処理の例
ファイル操作での例外処理
次のコードは安全にファイルの読み書きを行う関数です。try-except構文を使って例外処理を実装し、まずdata.txtにサンプルデータを書き込み、続いてその内容を読み取って表示します。書き込みや読み込み時にエラーが発生した場合は、IOError・FileNotFoundError・PermissionErrorを捕捉して適切なメッセージを表示し、安全に処理を終了します。
def safe_file_operations():
# ファイル書き込み
try:
with open("data.txt", "w") as file:
file.write("サンプルデータ\n")
except IOError as e:
print(f"ファイル書き込みエラー: {e}")
return False
# ファイル読み込み
try:
with open("data.txt", "r") as file:
content = file.read()
print("ファイル内容:", content)
except FileNotFoundError:
print("ファイルが見つかりません")
except PermissionError:
print("ファイルへのアクセス権限がありません")
except IOError as e:
print(f"ファイル読み込みエラー: {e}")
return True
safe_file_operations()
ユーザー入力の検証
while Trueで繰り返し入力を促し、int()で数値変換を試みます。0以下の数値や整数以外が入力された場合はValueErrorを発生させ、エラーメッセージを表示して再入力を求めます。Ctrl+Cによる中断(KeyboardInterrupt)も検知し、丁寧に終了メッセージを出して安全にプログラムを終了します。
def get_positive_integer():
while True:
try:
user_input = input("正の整数を入力してください: ")
number = int(user_input)
if number <= 0:
raise ValueError("正の整数を入力してください")
return number
except ValueError as e:
print(f"無効な入力: {e}")
except KeyboardInterrupt:
print("\nプログラムが中断されました")
return None
# ユーザー入力の取得
number = get_positive_integer()
if number is not None:
print(f"入力された数値: {number}")
具体的な例外の捕捉
可能な限り具体的な例外タイプを捕捉しましょう。上段では、FileNotFoundErrorやValueErrorなど特定の例外だけを捕捉して原因を明確にし、安全に処理を行う方法を示しています。一方、下段のexcept:のようにすべての例外を一括で捕捉するのは非推奨で、予期せぬエラーを隠してデバッグを難しくするため注意が必要です。
# 推奨される方法
try:
# 何らかの処理
pass
except FileNotFoundError:
# ファイルが見つからない場合の処理
pass
except ValueError:
# 値が不正な場合の処理
pass
# 非推奨な方法(あまりに広範な例外捕捉)
try:
# 何らかの処理
pass
except: # すべての例外を捕捉
pass
例外の再送出
場合によっては、例外を捕捉した後、再送出することが適切な場合があります。このコードは、例外の再送出(re-raise)と階層的なエラーハンドリングを示す例です。process_data()関数では、データを整数に変換しようとしますが、失敗した場合にValueErrorを表示した後、raiseで例外を再送出します。これにより上位のhigher_level_function()が例外を受け取り、再度処理を行います。最終的に高レベル関数でエラーメッセージを表示し、Noneを返して安全に終了します。
def process_data(data):
try:
# データ処理
result = int(data)
return result
except ValueError:
print("データの変換に失敗しました")
raise # 例外を再送出
def higher_level_function(data):
try:
return process_data(data)
except ValueError:
print("高レベルでのエラー処理")
return None
コンテキストマネージャの利用
リソース管理にはwith文を使用しましょう。このコードは、ファイルの安全な開閉方法を比較しています。上の「良い例」ではwith構文を使い、処理終了時にファイルが自動的に閉じられるため、エラーが起きてもリソースリークを防げます。一方、下の「悪い例」ではopen()とfinallyで明示的に閉じていますが、コードが冗長でエラー処理も複雑になります。したがって、Pythonではwith構文の使用が推奨されます。
# 良い例
with open("file.txt", "r") as file:
content = file.read()
# ファイルは自動的に閉じられる
# 悪い例
file = open("file.txt", "r")
try:
content = file.read()
finally:
file.close()
まとめ
例外処理は、堅牢で信頼性の高いプログラムを作成するために不可欠な技術です。try-except構文を適切に使用することで、予期しないエラー状況でもプログラムを適切に制御し、ユーザーに分かりやすいフィードバックを提供できます。
基本的な例外捕捉から、else節・finally節の活用、具体的な例外タイプの識別まで、段階的に理解を深めることが重要です。また、例外処理のベストプラクティスに従うことで、メンテナンス性の高いコードを書くことができます。
次のセクションでは、独自例外の定義方法について学び、より詳細で意味のあるエラーハンドリングを実現する方法を探求します。
演習問題
初級問題
問題1: 基本的な例外処理
整数の除算を行う関数を作成してください。ゼロ除算が発生した場合、"ゼロで割ることはできません"というメッセージを表示し、Noneを返すようにしてください。
問題2: 型エラーの処理
2つの数値の加算を行う関数を作成してください。数値以外が渡された場合、"数値型のみ処理できます"というメッセージを表示し、Noneを返すようにしてください。
問題3: ファイル操作の例外処理
指定されたファイル名でファイルを開き、内容を表示する関数を作成してください。ファイルが存在しない場合は"ファイルが見つかりません"、アクセス権限がない場合は"ファイルにアクセスできません"というメッセージを表示してください。
中級問題
問題4: 複数例外の処理
ユーザーから入力を受け取り、整数に変換して逆数を計算するプログラムを作成してください。値エラーとゼロ除算の両方を処理し、適切なエラーメッセージを表示してください。
問題5: else節の活用
リストのインデックスアクセスを行う関数を作成してください。インデックスが範囲外の場合は例外を処理し、正常な場合はelse節でアクセスした要素を表示してください。
問題6: finally節の使用
ファイル書き込みを行う関数を作成し、成功・失敗に関わらず"処理完了"というメッセージをfinally節で表示するようにしてください。
問題7: 例外情報の取得
除算を行う関数で、例外発生時に例外オブジェクトからエラーメッセージを取得して表示するようにしてください。
問題8: 例外の階層
さまざまな例外を発生させるコードを書き、例外の階層を理解するために、具体的な例外とその親クラス例外の両方を捕捉する処理を書いてください。
問題9: 入力検証ループ
ユーザーに数値の入力を求め、有効な数値が入力されるまで繰り返しプロンプトを表示するプログラムを作成してください。キーボード割り込み(Ctrl+C)も処理できるようにしてください。
上級問題
問題10: コンテキストマネージャの実装
独自のコンテキストマネージャクラスを作成し、with文で使用できるようにしてください。例外が発生しても確実にリソースが解放されることを確認してください。
問題11: 例外の再送出と変換
低レベルの例外を捕捉し、より高レベルの独自例外に変換して再送出するコードを作成してください。
問題12: 総合的な例外処理システム
ファイルの読み込み、データ処理、結果の保存を行うプログラムを作成し、各段階で発生する可能性のある例外を包括的に処理するシステムを実装してください。適切なロギングとリソース管理を含めてください。
初級問題
問題1: 基本的な例外処理
def safe_divide(a, b):
"""
整数の除算を行う関数
ゼロ除算が発生した場合、エラーメッセージを表示しNoneを返す
"""
try:
result = a / b
return result
except ZeroDivisionError:
print("ゼロで割ることはできません")
return None
# テスト
print("=== 問題1テスト ===")
print(safe_divide(10, 2)) # 5.0
print(safe_divide(10, 0)) # エラーメッセージ表示後、None
問題2: 型エラーの処理
def add_numbers(a, b):
"""
2つの数値の加算を行う関数
数値以外が渡された場合、エラーメッセージを表示しNoneを返す
"""
try:
result = a + b
return result
except TypeError:
print("数値型のみ処理できます")
return None
# テスト
print("\n=== 問題2テスト ===")
print(add_numbers(5, 3)) # 8
print(add_numbers(5, "3")) # エラーメッセージ表示後、None
print(add_numbers("5", 3)) # エラーメッセージ表示後、None
問題3: ファイル操作の例外処理
def display_file_content(filename):
"""
指定されたファイルの内容を表示する関数
ファイルが存在しない場合やアクセス権限がない場合に適切なメッセージを表示
"""
try:
with open(filename, 'r', encoding='utf-8') as file:
content = file.read()
print(f"ファイル内容:\n{content}")
return content
except FileNotFoundError:
print(f"ファイル '{filename}' が見つかりません")
except PermissionError:
print(f"ファイル '{filename}' にアクセスできません")
except Exception as e:
print(f"予期しないエラーが発生しました: {e}")
return None
# テスト
print("\n=== 問題3テスト ===")
# 存在するファイルでテスト(テスト用にファイルを作成)
with open("test_file.txt", "w", encoding='utf-8') as f:
f.write("これはテストファイルです\nHello, World!")
display_file_content("test_file.txt") # 正常に表示
display_file_content("nonexistent.txt") # ファイルが見つかりません
中級問題
問題4: 複数例外の処理
def calculate_reciprocal():
"""
ユーザー入力から整数の逆数を計算する
値エラーとゼロ除算を処理
"""
try:
user_input = input("整数を入力してください: ")
number = int(user_input)
reciprocal = 1 / number
print(f"{number} の逆数は {reciprocal} です")
return reciprocal
except ValueError:
print("エラー: 有効な整数を入力してください")
except ZeroDivisionError:
print("エラー: ゼロの逆数は計算できません")
return None
# テスト
print("\n=== 問題4テスト ===")
# テスト実行(手動で様々な入力でテスト)
# calculate_reciprocal()
問題5: else節の活用
def safe_list_access(lst, index):
"""
リストの安全なインデックスアクセス
正常な場合はelse節で要素を表示
"""
try:
element = lst[index]
except IndexError:
print(f"エラー: インデックス {index} は範囲外です")
return None
except TypeError:
print("エラー: インデックスは整数で指定してください")
return None
else:
print(f"インデックス {index} の要素: {element}")
return element
# テスト
print("\n=== 問題5テスト ===")
test_list = [10, 20, 30, 40, 50]
safe_list_access(test_list, 2) # 正常: 30
safe_list_access(test_list, 10) # 範囲外エラー
safe_list_access(test_list, "2") # 型エラー
問題6: finally節の使用
def write_to_file(filename, content):
"""
ファイル書き込みを行う関数
成功・失敗に関わらずfinally節でメッセージを表示
"""
file = None
try:
file = open(filename, 'w', encoding='utf-8')
file.write(content)
print("ファイルへの書き込みに成功しました")
return True
except IOError as e:
print(f"ファイル書き込みエラー: {e}")
return False
finally:
if file is not None:
file.close()
print("処理完了")
# テスト
print("\n=== 問題6テスト ===")
write_to_file("output.txt", "Hello, World!") # 成功ケース
# 権限のないディレクトリなどでテストすると失敗ケースを確認可能
問題7: 例外情報の取得
def detailed_division(a, b):
"""
除算を行う関数
例外発生時に詳細なエラー情報を表示
"""
try:
result = a / b
return result
except Exception as e:
print(f"エラーが発生しました: {e}")
print(f"エラーの型: {type(e).__name__}")
print(f"エラー引数: {e.args}")
return None
# テスト
print("\n=== 問題7テスト ===")
detailed_division(10, 2) # 5.0
detailed_division(10, 0) # 詳細なエラー情報を表示
detailed_division(10, "2") # 詳細なエラー情報を表示
問題8: 例外の階層
def exception_hierarchy_demo():
"""
例外の階層を理解するデモ
具体的な例外とその親クラス例外の両方を捕捉
"""
test_cases = [
lambda: 1 / 0, # ZeroDivisionError
lambda: int("abc"), # ValueError
lambda: [1, 2, 3][10], # IndexError
lambda: {}["key"], # KeyError
]
for i, case in enumerate(test_cases):
print(f"\nテストケース {i+1}:")
try:
case()
except (ZeroDivisionError, ValueError) as e:
print(f"特定の例外を捕捉: {type(e).__name__}: {e}")
except Exception as e:
print(f"一般的な例外を捕捉: {type(e).__name__}: {e}")
else:
print("例外は発生しませんでした")
# テスト
print("\n=== 問題8テスト ===")
exception_hierarchy_demo()
問題9: 入力検証ループ
def get_valid_number():
"""
有効な数値が入力されるまで繰り返しプロンプトを表示
キーボード割り込みも処理
"""
while True:
try:
user_input = input("数値を入力してください (終了するには 'quit' と入力): ")
if user_input.lower() == 'quit':
print("プログラムを終了します")
return None
number = float(user_input)
print(f"入力された数値: {number}")
return number
except ValueError:
print("エラー: 有効な数値を入力してください")
except KeyboardInterrupt:
print("\nプログラムがユーザーによって中断されました")
return None
# テスト
print("\n=== 問題9テスト ===")
# get_valid_number() # コメントを外してテスト実行
上級問題
問題10: コンテキストマネージャの実装
class DatabaseConnection:
"""
データベース接続を模したコンテキストマネージャ
"""
def __init__(self, database_name):
self.database_name = database_name
self.connected = False
def __enter__(self):
print(f"データベース '{self.database_name}' に接続中...")
# 実際のデータベース接続処理を模擬
self.connected = True
print("接続成功")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("データベース接続を閉じています...")
self.connected = False
if exc_type is not None:
print(f"例外が発生しました: {exc_type.__name__}: {exc_val}")
print("接続終了")
return True # 例外を再送出しない
def execute_query(self, query):
"""クエリ実行を模擬"""
if not self.connected:
raise RuntimeError("データベースに接続されていません")
print(f"クエリを実行: {query}")
# 実際のクエリ実行を模擬
return f"'{query}' の結果"
# テスト
print("\n=== 問題10テスト ===")
# 正常な使用例
print("--- 正常な使用例 ---")
with DatabaseConnection("my_database") as db:
result = db.execute_query("SELECT * FROM users")
print(result)
print("\n--- 例外が発生するケース ---")
# 例外が発生するケース
with DatabaseConnection("my_database") as db:
result = db.execute_query("SELECT * FROM products")
print(result)
# 意図的に例外を発生
raise ValueError("テスト例外")
問題11: 例外の再送出と変換
# カスタム例外の定義
class DataProcessingError(Exception):
"""データ処理中のエラーを表すカスタム例外"""
pass
class ValidationError(DataProcessingError):
"""データ検証エラー"""
pass
class CalculationError(DataProcessingError):
"""計算エラー"""
pass
def low_level_processing(data):
"""
低レベルデータ処理
様々な例外を発生させる可能性がある
"""
if not isinstance(data, (int, float)):
raise TypeError("データは数値である必要があります")
if data < 0:
raise ValueError("データは正の値である必要があります")
if data == 0:
raise ZeroDivisionError("ゼロ除算が発生します")
return 100 / data
def high_level_processing(data_list):
"""
高レベルデータ処理
低レベルの例外を捕捉し、高レベルの例外に変換
"""
results = []
for i, data in enumerate(data_list):
try:
result = low_level_processing(data)
results.append(result)
except (TypeError, ValueError) as e:
# 検証エラーに変換
raise ValidationError(f"データ {data} (インデックス {i}) が無効です: {e}")
except ZeroDivisionError as e:
# 計算エラーに変換
raise CalculationError(f"データ {data} (インデックス {i}) で計算エラー: {e}")
except Exception as e:
# その他の例外をデータ処理エラーに変換
raise DataProcessingError(f"予期しないエラー: {e}")
return results
# テスト
print("\n=== 問題11テスト ===")
# 正常なデータ
try:
results = high_level_processing([10, 20, 30])
print(f"処理結果: {results}")
except DataProcessingError as e:
print(f"データ処理エラー: {e}")
# 無効なデータ
try:
results = high_level_processing([10, -5, 30])
print(f"処理結果: {results}")
except DataProcessingError as e:
print(f"データ処理エラー: {e}")
# ゼロ除算
try:
results = high_level_processing([10, 0, 30])
print(f"処理結果: {results}")
except DataProcessingError as e:
print(f"データ処理エラー: {e}")
問題12: 総合的な例外処理システム
import logging
import datetime
# ロギングの設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('data_processing.log', encoding='utf-8'),
logging.StreamHandler()
]
)
class DataProcessor:
"""
ファイル読み込み、データ処理、結果保存を行う総合的な処理システム
"""
def __init__(self):
self.logger = logging.getLogger(__name__)
def read_data_file(self, filename):
"""データファイルの読み込み"""
self.logger.info(f"ファイル読み込み開始: {filename}")
try:
with open(filename, 'r', encoding='utf-8') as file:
lines = file.readlines()
# データの前処理
data = []
for i, line in enumerate(lines):
line = line.strip()
if line: # 空行でない場合
try:
number = float(line)
data.append(number)
except ValueError as e:
self.logger.warning(f"行 {i+1}: 数値に変換できないデータ '{line}' をスキップします")
self.logger.info(f"ファイル読み込み完了: {len(data)} 個の有効なデータを読み込み")
return data
except FileNotFoundError:
self.logger.error(f"ファイルが見つかりません: {filename}")
raise
except PermissionError:
self.logger.error(f"ファイルアクセス権限がありません: {filename}")
raise
except Exception as e:
self.logger.error(f"ファイル読み込み中に予期しないエラー: {e}")
raise
def process_data(self, data):
"""データ処理"""
self.logger.info("データ処理開始")
try:
if not data:
raise ValueError("処理するデータがありません")
# 統計計算
results = {
'count': len(data),
'sum': sum(data),
'average': sum(data) / len(data),
'max': max(data),
'min': min(data),
'processed_at': datetime.datetime.now().isoformat()
}
self.logger.info("データ処理完了")
return results
except ZeroDivisionError:
self.logger.error("データ数がゼロのため平均値を計算できません")
raise
except Exception as e:
self.logger.error(f"データ処理中にエラー: {e}")
raise
def save_results(self, results, output_filename):
"""結果の保存"""
self.logger.info(f"結果保存開始: {output_filename}")
try:
with open(output_filename, 'w', encoding='utf-8') as file:
file.write("データ処理結果\n")
file.write("=" * 50 + "\n")
for key, value in results.items():
file.write(f"{key}: {value}\n")
self.logger.info("結果保存完了")
return True
except IOError as e:
self.logger.error(f"結果保存中にエラー: {e}")
raise
def run_pipeline(self, input_file, output_file):
"""
完全なデータ処理パイプラインの実行
"""
self.logger.info("=== データ処理パイプライン開始 ===")
data = None
try:
# ステップ1: ファイル読み込み
data = self.read_data_file(input_file)
# ステップ2: データ処理
results = self.process_data(data)
# ステップ3: 結果保存
self.save_results(results, output_file)
self.logger.info("=== データ処理パイプライン正常終了 ===")
return True
except Exception as e:
self.logger.error(f"=== データ処理パイプライン異常終了: {e} ===")
return False
finally:
# リソースクリーンアップ(必要な場合)
self.logger.info("リソースクリーンアップ完了")
# テスト
print("\n=== 問題12テスト ===")
# テストデータの作成
with open("test_data.txt", "w", encoding='utf-8') as f:
f.write("10.5\n")
f.write("20.3\n")
f.write("15.7\n")
f.write("abc\n") # 無効なデータ
f.write("25.1\n")
f.write("0\n") # ゼロ
f.write("30.9\n")
# データ処理の実行
processor = DataProcessor()
success = processor.run_pipeline("test_data.txt", "results.txt")
if success:
print("データ処理が正常に完了しました")
# 結果ファイルの内容を表示
with open("results.txt", "r", encoding='utf-8') as f:
print("\n処理結果:")
print(f.read())
else:
print("データ処理中にエラーが発生しました")
print("\n=== すべての演習問題が完了しました ===")