家計簿アプリを作成する③【SQLAlchemyのEngineを作成】

Python

家計簿アプリ作成【SQLAlchemyのEngineを作成する】

SQLAlchemyでデータベースへの接続とSQL操作の実行を管理するEngineクラスを作成する

Engineモジュール

Engineクラス

Engineクラスは、データベースへの接続とSQL操作の実行を行う

家計簿アプリを作成する②で作成した各テーブル情報に接続する各アダプタークラスを作成し、接続する設計にする
Engineクラスに全テーブルの処理を記載するより、Engineクラスを継承したアダプタークラスを作成することによって、異なるテーブルから情報を取得するバグを軽減できると考えたため

Engineクラス内では以下の自作したクラスを利用する
Message
Create_Exception_Log
Const
・Message_Box

各自作クラスは下に記載する

【ログ作成】
・引数で渡された情報を元にlogテーブルにログをINSERTする

log_seqlog_kindsfunction_idlog_detailuser_idoperate_at
510SI-SEL-USERSログイン画面 users:test 取得 件login2025-06-13 20:27:52
511WI-SEL-USERSTypeError(“object of type ‘User’ has no len()”)login2025-06-13 20:27:52
512SI-SEL-USERSログイン画面 users:test 取得 件login2025-06-13 20:45:01
513EI-SEL-USERSログイン画面 users:test 取得 1件 認証失敗login2025-06-13 20:45:02

・ログ出力時に例外エラーが発生した場合、ログ出力時例外エラーファイルを作成するログ出力時例外エラーメッセージを作成する
ログ出力時例外エラーメッセージ:データベース操作時にエラーが発生しました。管理者に連絡してください。
・ログ出力時例外エラーファイル作成時に例外エラーが発生した場合、例外エラーファイル出力時の例外エラー用メッセージを作成する
ログ出力時例外エラーメッセージ:ログファイル作成時にエラーが発生しました。管理者に連絡してください。

【例外エラー用メッセージ作成】
DB接続処理中に例外エラーが発生した場合
例外エラー用メッセージを作成する
例外エラー用メッセージ:データベース操作時にエラーが発生しました。管理者に連絡してください。
終了ログを作成する

from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import Session
from db.models import Log
from common.message import Message
from common.create_exception_log import Create_Exception_Log
from common.const import Const
from common.message_box import Message_Box


# DB接続
class Engine:
    def __init__(self):
        self.engine = create_engine("sqlite:///db/example.db", echo=True)
        self.inspector = inspect(self.engine)
        # セッションクラス
        self.session = Session(self.engine, expire_on_commit=False)
        # ログクラス
        self.log_row = Log()
        # メッセージクラス
        self.message = Message()
        # 例外エラー出力クラス
        self.create_exception_log = Create_Exception_Log()
        # 定数クラス
        self.const = Const()

    # ログ作成
    def create_log(
        self, arg_log_kinds, arg_log_function_id, arg_log_detail, arg_entry_user_id
    ):
        """
        ログを作成する
        引数
            arg_log_kinds:ログ種類,
            arg_log_function_id:処理ID,
            arg_log_detail:ログ詳細,
            arg_entry_user_id:操作ユーザー
        """
        error_message = None
        # 引数を元にログクラスをインスタンス
        try:
            log = Log(
                log_kinds=arg_log_kinds,
                function_id=arg_log_function_id,
                log_detail=arg_log_detail,
                user_id=arg_entry_user_id,
            )
            # インスタンスしたログをセッションに追加
            self.session.add(log)
            # データベースに反映
            self.session.commit()
            return error_message
        except Exception as e:
            # 例外エラーテキストファイルを出力
            isError = self.create_exception_log.create_exception_log(
                arg_log_function_id, e
            )
            if isError == self.const.Log_Error_Kbn.LOGFILE_CREATE_ERROR:
                # 例外エラーテキストファイルを出力失敗時、エラー
                error_message = self.exception_log_exception()
            elif isError == self.const.Log_Error_Kbn.LOG_EXCEPTION_ERROR:
                # 例外エラーテキストファイルを出力成功時、エラー
                error_message = Message_Box()
                # インスタンスしたメッセージボックスにidとメッセージを代入
                error_message.message_id = "HAB002W"
                error_message.message_text = self.message.Message_Box.HAB002W
            # 例外エラーメッセージを返す
            return error_message
        finally:
            # セッションを閉じる
            self.session.close()

    # ログ出力時例外エラー用メッセージ作成
    def exception_log_exception(self):
        """
        ログ出力時の例外エラー用メッセージを作成する
        """
        # メッセージボックスクラスをインスタンス
        message_box = Message_Box()
        # インスタンスしたメッセージボックスにidとメッセージを代入
        message_box.message_id = "HAB001W"
        message_box.message_text = self.message.Message_Box.HAB001W
        # インスタンスしたメッセージボックスを返す
        return message_box

    # 例外エラー用メッセージ作成、ログ作成
    def exception_log(self, arg_log_function_id, arg_e, arg_entry_user_id):
        """
        例外エラー用メッセージ作成、ログ作成
        引数
            arg_log_kinds:ログ種類,
            arg_log_function_id:処理ID,
            arg_log_detail:ログ詳細,
            arg_entry_user_id:操作ユーザー
        """
        # メッセージボックスクラスをインスタンス
        message_box = Message_Box()
        # インスタンスしたメッセージボックスにidとメッセージを代入
        message_box.message_id = "HAB002W"
        message_box.message_text = self.message.Message_Box.HAB002W
        # ログ作成
        self.create_log(
            Const.Log_Kinds.WARNING,
            arg_log_function_id,
            repr(arg_e),
            arg_entry_user_id,
        )
        # インスタンスしたメッセージボックスを返す
        return message_box

Return_Infoクラス

Return_Infoクラスは、アダプタークラスで取得した情報と画面のメッセージボックスに表示する情報を保持する

# 返却用
class Return_Info:
    def __init__(self):
        # 取得した情報を代入する
        self.return_row = None
        # 画面のメッセージボックスに表示する情報を代入する
        self.return_message_box = Message_Box()

Messageクラス

Messageクラスは、処理ID、ログ詳細、画面のメッセージボックスなどで利用するメッセージ一覧を保持する

# メッセージ一覧
class Message:

    # 処理ID用
    class Log_Function_Id:
        id = "{0}-{1}-{2}"

    # ログ詳細用
    class Log_Message:
        LOGIN = "ログイン画面 users:{0} 取得 {1}件"
        LOGIN_SUCCESS = "ログイン画面 users:{0} 取得 {1}件 認証成功"
        LOGIN_ERROR = "ログイン画面 users:{0} 取得 {1}件 認証失敗"
        FILL = "{0}画面 {1}{2} 取得 {3}件"
        Fill_NO_ROW = "{0}画面 {1}{2} 取得 0件"
        INSERT = "{0}画面 {1}{2} 追加"
        NON_INSERT = "{0}画面 {1}{2} 追加 ID重複"
        UPDATE = "{0}画面 {1}{2} 更新"
        NON_UPDATE = "{0}画面 {1}{2} 更新 排他エラー"
        DELETE = "{0}画面 {1}{2} 削除"
        NON_DELETE = "{0}画面 {1}{2} 削除 排他エラー"

    # メッセージボックス用
    class Message_Box:
        # INFO
        HAB001I = "新規登録が完了しました。"
        HAB002I = "更新が完了しました。"
        HAB003I = "削除が完了しました。"
        # CAUTION
        HAB001C = "削除されたため表示できません。"
        HAB002C = "同じユーザーIDのユーザーが存在します。\nユーザーIDを変更して登録してください。"
        HAB003C = "他のユーザーが更新したため、更新に失敗しました。\n画面を再度開きなおしてください。"
        HAB004C = "{0}に失敗しました。"
        HAB005C = "{0}を入力してください。"
        # WARNING
        HAB001W = (
            "ログファイル作成時にエラーが発生しました。\n管理者に連絡してください。"
        )
        HAB002W = (
            "データベース操作時にエラーが発生しました。\n管理者に連絡してください。"
        )

Create_Exception_Logクラス

Create_Exception_Logクラスは、ログ作成で例外エラーが発生した場合、規定フォルダにテキストファイルを出力する

from common.const import Const
from datetime import datetime
from zoneinfo import ZoneInfo


# ログ出力時例外エラーテキストファイル出力
class Create_Exception_Log:
    # テキストファイル出力
    def create_exception_log(self, arg_function_id, arg_log_detail):
        """
        ログ出力例外エラー用テキストファイル出力
        引数
            arg_function_id:処理ID,
            arg_log_detail:例外エラー内容
        """
        f = None
        try:
            # 規定フォルダにファイルがない場合は作成して開く
            # ファイルがある場合はカーソルが最下行の状態で開く
            f = open(
                f"{Const.FilePath.EXCEPTION_FOLDER}/{Const.FilePath.EXCEPTION_FILE}",
                "a+",
                encoding="utf-888",
            )
            # 「現在時間   処理ID  例外エラー内容」を書き込む
            f.write(
                f"{datetime.now(ZoneInfo("Asia/Tokyo"))}  {arg_function_id}   {arg_log_detail}\n"
            )
            return Const.Log_Error_Kbn.LOG_EXCEPTION_ERROR
        except:
            return Const.Log_Error_Kbn.LOGFILE_CREATE_ERROR
        finally:
            # ファイルが開かれている場合
            if f is not None:
                # ファイルを閉じる
                f.close()

Constクラス

Constクラスは、定数文字を保持する

# 定数
class Const:
    # エラー区分
    class Log_Error_Kbn:
        LOGFILE_CREATE_ERROR = 8
        LOG_EXCEPTION_ERROR = 9

    # ファイルパス
    class FilePath:
        EXCEPTION_FOLDER = "log"
        EXCEPTION_FILE = "exception_log.txt"

    # ログ用種類
    class Log_Kinds:
        START = "S"
        END = "E"
        INFO = "I"
        CAUTION = "C"
        WARNING = "W"

    # ログ用処理
    class Log_Process:
        SELECT = "SEL"
        INSERT = "INS"
        UPDATE = "UPD"
        DELETE = "DEL"

    # ログ用機能
    class Log_Function:
        HAB = "HAB"
        USERS = "USERS"

    # 文字
    class Const_Text:
        TEXT_BLANK = ""

Message_Boxクラス

Message_Boxクラスは、画面のメッセージボックスに表示する情報を保持する

# 画面用メッセージボックス
class Message_Box:
    def __init__(self):
        # 画面のメッセージボックスに表示するIDを代入する
        self.message_id = None
        # 画面のメッセージボックスに表示するメッセージを代入する
        self.message_text = None

全体コード(GitHub)

household_account_book/db/common/engine.py at main · SakumaTakayuki/household_account_book
Contribute to SakumaTakayuki/household_account_book development by creating an account on GitHub.