Pythonの例外処理を使ったエラーハンドリングの具体例6選

try:
    # 例外が発生する可能性のあるコード
except ExceptionType:
    # 例外が発生した場合の処理

try:
    num = int(input("整数を入力してください: "))
    result = 10 / num
    print(f"10 ÷ {num} = {result}")
except ValueError:
    print("整数を入力してください。")
except ZeroDivisionError:
    print("0では割り算できません。")

def read_file(filename):
    try:
        with open(filename, 'r') as file:
            content = file.read()
            print(content)
    except FileNotFoundError:
        print(f"ファイル '{filename}' が見つかりません。")
    except IOError:
        print(f"ファイル '{filename}' の読み込み中にエラーが発生しました。")

# 関数の使用例
read_file("example.txt")

def divide_numbers(a, b):
    try:
        result = a / b
        return result
    except ZeroDivisionError:
        print("ゼロで割ることはできません。")
    except TypeError:
        print("数値を入力してください。")
    except Exception as e:
        print(f"予期せぬエラーが発生しました: {e}")

# 関数の使用例
print(divide_numbers(10, 2))  # 正常な場合
print(divide_numbers(10, 0))  # ゼロ除算
print(divide_numbers("10", 2))  # 型エラー
print(divide_numbers(10, "2"))  # 型エラー

5.0
ゼロで割ることはできません。
None
数値を入力してください。
None
数値を入力してください。
None

def read_and_process_file(filename):
    try:
        with open(filename, 'r') as file:
            content = file.read()
            # ファイルの内容を処理する何らかの操作
            processed_content = content.upper()
    except FileNotFoundError:
        print(f"ファイル '{filename}' が見つかりません。")
        return None
    except IOError:
        print(f"ファイル '{filename}' の読み込み中にエラーが発生しました。")
        return None
    else:
        print("ファイルの読み込みと処理が成功しました。")
        return processed_content
    finally:
        print("ファイル操作の後処理を実行します。")

# 関数の使用例
result = read_and_process_file("example.txt")
if result:
    print(f"処理結果: {result[:50]}...")  # 最初の50文字だけ表示

class InvalidAgeError(Exception):
    def __init__(self, age, message="無効な年齢です。"):
        self.age = age
        self.message = message
        super().__init__(self.message)

def verify_age(age):
    if age < 0 or age > 150:
        raise InvalidAgeError(age, f"{age}は有効な年齢ではありません。0から150の間で入力してください。")
    print(f"年齢 {age} は有効です。")

# 関数の使用例
try:
    verify_age(30)  # 正常な場合
    verify_age(200)  # 無効な年齢
except InvalidAgeError as e:
    print(f"エラー: {e}")

年齢 30 は有効です。
エラー: 200は有効な年齢ではありません。0から150の間で入力してください。

import logging

logging.basicConfig(level=logging.INFO)

def divide(a, b):
    try:
        result = a / b
    except ZeroDivisionError:
        logging.error("ゼロ除算が発生しました")
        raise  # 例外を再送出
    return result

def calculate():
    try:
        result = divide(10, 0)
    except ZeroDivisionError:
        print("計算中にエラーが発生しました")

# 関数の使用例
calculate()

ERROR:root:ゼロ除算が発生しました
計算中にエラーが発生しました

class DatabaseConnection:
    def __init__(self, db_name):
        self.db_name = db_name

    def __enter__(self):
        print(f"データベース '{self.db_name}' に接続しています...")
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is not None:
            print(f"エラーが発生しました: {exc_type.__name__} - {exc_value}")
        print(f"データベース '{self.db_name}' との接続を閉じています...")
        return False  # 例外を伝播させる

    def query(self, sql):
        if "ERROR" in sql.upper():
            raise ValueError("不正なSQLクエリです")
        print(f"クエリを実行: {sql}")

# コンテキストマネージャの使用例
try:
    with DatabaseConnection("MyDB") as db:
        db.query("SELECT * FROM users")
        db.query("INSERT ERROR")  # エラーを発生させる
except ValueError as e:
    print(f"ValueError: {e}")

データベース 'MyDB' に接続しています...
クエリを実行: SELECT * FROM users
エラーが発生しました: ValueError - 不正なSQLクエリです
データベース 'MyDB' との接続を閉じています...
ValueError: 不正なSQLクエリです

import asyncio

async def fetch_data(url):
    print(f"{url} からデータを取得中...")
    await asyncio.sleep(2)  # ネットワーク遅延をシミュレート
    if "error" in url:
        raise ConnectionError(f"{url} からのデータ取得に失敗しました")
    return f"{url} からのデータ"

async def main():
    urls = [
        "https://api.example.com/data1",
        "https://api.example.com/error",
        "https://api.example.com/data2"
    ]
    tasks = [fetch_data(url) for url in urls]

    results = []
    for task in asyncio.as_completed(tasks):
        try:
            result = await task
            results.append(result)
        except ConnectionError as e:
            print(f"エラー: {e}")

    print("取得したデータ:")
    for result in results:
        print(result)

# 非同期関数の実行
asyncio.run(main())

https:// api.example.com/data1 からデータを取得中...
https:// api.example.com/error からデータを取得中...
https:// api.example.com/data2 からデータを取得中...
エラー: https:// api.example.com/error からのデータ取得に失敗しました
取得したデータ:
https:// api.example.com/data1 からのデータ
https:// api.example.com/data2 からのデータ

# リンクの都合上、リンクの箇所に意図的にスペースを入れています

def process_file(filename):
    try:
        with open(filename, 'r') as file:
            content = file.read()
            # ファイルの内容を処理する
            processed_content = content.upper()
            return processed_content
    except FileNotFoundError:
        print(f"ファイル '{filename}' が見つかりません。")
    except PermissionError:
        print(f"ファイル '{filename}' にアクセスする権限がありません。")
    except IOError as e:
        print(f"ファイル '{filename}' の読み込み中にエラーが発生しました: {e}")
    except Exception as e:
        print(f"予期せぬエラーが発生しました: {e}")

# 関数の使用例
result = process_file("example.txt")
if result:
    print("処理結果:", result[:50])  # 最初の50文字だけ表示

import time
import random

def connect_to_server(server_url, max_retries=3):
    for attempt in range(max_retries):
        try:
            # サーバーへの接続をシミュレート
            if random.random() < 0.7:  # 70%の確率で接続成功
                print(f"サーバー '{server_url}' に接続しました。")
                return True
            else:
                raise ConnectionError("接続に失敗しました。")
        except ConnectionError as e:
            print(f"試行 {attempt + 1}/{max_retries}: {e}")
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # 指数バックオフ
                print(f"{wait_time}秒待機してリトライします...")
                time.sleep(wait_time)

    print(f"サーバー '{server_url}' への接続に {max_retries} 回失敗しました。")
    return False

# 関数の使用例
success = connect_to_server("https://example.com")
if success:
    print("サーバーとの通信を開始します。")
else:
    print("サーバーとの接続を確立できませんでした。後でもう一度お試しください。")

import logging

# ログの設定
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def divide_numbers(a, b):
    try:
        result = a / b
        logging.info(f"除算成功: {a} / {b} = {result}")
        return result
    except ZeroDivisionError:
        logging.error(f"ゼロ除算エラー: {a} を 0 で割ろうとしました")
        raise
    except TypeError as e:
        logging.error(f"型エラー: {e}")
        raise

# 関数の使用例
try:
    print(divide_numbers(10, 2))
    print(divide_numbers(10, 0))
except Exception as e:
    print(f"エラーが発生しました: {e}")

try:
    print(divide_numbers("10", 2))
except Exception as e:
    print(f"エラーが発生しました: {e}")

2024-07-14 12:34:56,789 - INFO - 除算成功: 10 / 2 = 5.0
5.0
2024-07-14 12:34:56,790 - ERROR - ゼロ除算エラー: 10 を 0 で割ろうとしました
エラーが発生しました: division by zero
2024-07-14 12:34:56,791 - ERROR - 型エラー: unsupported operand type(s) for /: 'str' and 'int'
エラーが発生しました: unsupported operand type(s) for /: 'str' and 'int'

def greet(name):
    message = f"こんにちは、{name}さん！"
    print(mesage)  # タイプミス: 'message' のつもりが 'mesage' になっている

greet("太郎")

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 3, in greet
NameError: name 'mesage' is not defined

def greet(name):
    message = f"こんにちは、{name}さん！"
    print(message)  # 'mesage' を 'message' に修正

greet("太郎")

def calculate_age(birth_year):
    current_year = 2024
    age = current_year - birth_year
    return "あなたは" + age + "歳です。"

print(calculate_age(1990))

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 4, in calculate_age
TypeError: can only concatenate str (not "int") to str

def calculate_age(birth_year):
    current_year = 2024
    age = current_year - birth_year
    return "あなたは" + str(age) + "歳です。"

print(calculate_age(1990))

def double_number():
    user_input = input("数字を入力してください: ")
    number = int(user_input)
    result = number * 2
    print(f"{number}の2倍は{result}です。")

double_number()

数字を入力してください: abc
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 3, in double_number
ValueError: invalid literal for int() with base 10: 'abc'

def double_number():
    while True:
        try:
            user_input = input("数字を入力してください: ")
            number = int(user_input)
            result = number * 2
            print(f"{number}の2倍は{result}です。")
            break
        except ValueError:
            print("無効な入力です。数字を入力してください。")

double_number()

数字を入力してください: abc
無効な入力です。数字を入力してください。
数字を入力してください: 5
5の2倍は10です。

import os

def read_and_write_file(input_file, output_file):
    try:
        # 入力ファイルを読み込む
        with open(input_file, 'r') as file:
            content = file.read()

        # 内容を加工する（ここでは単純に大文字に変換）
        processed_content = content.upper()

        # 出力ファイルに書き込む
        with open(output_file, 'w') as file:
            file.write(processed_content)

        print(f"ファイルの処理が完了しました。結果は {output_file} に保存されています。")

    except FileNotFoundError:
        print(f"エラー: ファイル '{input_file}' が見つかりません。")
    except PermissionError:
        print(f"エラー: ファイル '{input_file}' または '{output_file}' にアクセスする権限がありません。")
    except IOError as e:
        print(f"エラー: ファイル操作中に問題が発生しました: {e}")
    except Exception as e:
        print(f"予期せぬエラーが発生しました: {e}")
    finally:
        print("ファイル操作を終了します。")

# 関数の使用例
read_and_write_file("input.txt", "output.txt")

ファイルの処理が完了しました。結果は output.txt に保存されています。
ファイル操作を終了します。

エラー: ファイル 'input.txt' が見つかりません。
ファイル操作を終了します。

import requests
import time

def fetch_data_from_api(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()  # HTTPエラーがあれば例外を発生させる
            return response.json()

        except requests.exceptions.HTTPError as e:
            print(f"HTTPエラー: {e}")
            if response.status_code == 429:  # レートリミット
                print("レートリミットに達しました。しばらく待機します。")
                time.sleep(60)  # 1分待機
            else:
                print(f"リトライ {attempt + 1}/{max_retries}")

        except requests.exceptions.ConnectionError:
            print("接続エラー: サーバーに接続できません。")

        except requests.exceptions.Timeout:
            print("タイムアウト: サーバーからの応答がありません。")

        except requests.exceptions.RequestException as e:
            print(f"リクエスト中にエラーが発生しました: {e}")

        if attempt < max_retries - 1:
            wait_time = 2 ** attempt  # 指数バックオフ
            print(f"{wait_time}秒待機してリトライします...")
            time.sleep(wait_time)

    print(f"{max_retries}回のリトライ後も失敗しました。")
    return None

# 関数の使用例
api_url = "https://api.example.com/data"
data = fetch_data_from_api(api_url)
if data:
    print("取得したデータ:", data)
else:
    print("データの取得に失敗しました。")

接続エラー: サーバーに接続できません。
2秒待機してリトライします...
接続エラー: サーバーに接続できません。
4秒待機してリトライします...
接続エラー: サーバーに接続できません。
3回のリトライ後も失敗しました。
データの取得に失敗しました。

import sqlite3

def insert_user_data(db_file, user_id, name, email):
    connection = None
    try:
        connection = sqlite3.connect(db_file)
        cursor = connection.cursor()

        # ユーザーデータを挿入するSQLクエリ
        insert_query = """
        INSERT INTO users (user_id, name, email)
        VALUES (?, ?, ?)
        """
        cursor.execute(insert_query, (user_id, name, email))

        connection.commit()
        print(f"ユーザー '{name}' のデータを正常に挿入しました。")

    except sqlite3.IntegrityError:
        print(f"エラー: ユーザーID '{user_id}' は既に存在します。")
        connection.rollback()

    except sqlite3.OperationalError as e:
        print(f"データベース操作エラー: {e}")
        connection.rollback()

    except Exception as e:
        print(f"予期せぬエラーが発生しました: {e}")
        if connection:
            connection.rollback()

    finally:
        if connection:
            connection.close()
            print("データベース接続を閉じました。")

# 関数の使用例
db_file = "users.db"
insert_user_data(db_file, 1, "山田太郎", "yamada@example.com")
insert_user_data(db_file, 1, "鈴木花子", "suzuki@example.com")  # 重複エラーを発生させる

ユーザー '山田太郎' のデータを正常に挿入しました。
データベース接続を閉じました。
エラー: ユーザーID '1' は既に存在します。
データベース接続を閉じました。