DBユーティリティ

db_access.py

# coding:utf-8
import logging
import os
import sqlite3

from app.util import message_access, app_config

"""
DBアクセスを制御する
"""
__author__ = "t.ebinuma"
__version__ = "1.2"
__date__ = "28 April 2018"


def connect():
    """
    SQLコネクションを取得する
    """
    # 設定ファイルのDBファイルを指定する
    connection = sqlite3.connect(os.path.join(app_config.get_value('db_path'), app_config.get_value('db_file')))
    # 項目名でデータ取得できる設定を行う(sqlite3固有の設定です)
    connection.row_factory = sqlite3.Row
    # IsolationLevelをDEFERREDにする
    connection.isolation_level = 'DEFERRED'
    return connection


def close(connection):
    """
    SQLコネクションを閉じる
    """
    connection.close()


def fetch(sql, param):
    """
    fetchを行う(selectで使用する)
    """
    conn = connect()
    cursor = conn.cursor()
    logger = logging.getLogger(__name__)
    try:
        stmt = " \n".join(sql)
        logger.debug("sql={}".format(" ".join(sql)))
        logger.debug("param={}".format(param))
        cursor.execute(stmt, param)
        fetch_list = cursor.fetchall()
    except Exception as e:
        logger.exception(message_access.get_message('E900'), e)
        raise e
    finally:
        conn.close()
    return fetch_list


def execute(sql, param):
    """
    executeを行う(insertで使用する)
    """
    conn = connect()
    cursor = conn.cursor()
    logger = logging.getLogger(__name__)
    try:
        stmt = " \n".join(sql)
        logger = logging.getLogger(__name__)
        logger.debug("sql={}".format(" ".join(sql)))
        logger.debug("param={}".format(param))
        cursor.execute(stmt, param)
        row_id = cursor.lastrowid
        conn.commit()
    except Exception as e:
        conn.rollback()
        logger.exception(message_access.get_message('E900'), e)
        raise e
    finally:
        cursor.close()
        conn.close()
    return row_id


def ddl_execute(sql):
    """
    executeを行う(Drop Table等のDDLで使用する)
    """
    conn = connect()
    cursor = conn.cursor()
    logger = logging.getLogger(__name__)
    try:
        logger.debug("sql={}".format(sql))
        cursor.execute(sql)
    except Exception as e:
        logger.exception(message_access.get_message('E900'), e)
        raise e
    finally:
        cursor.close()
        conn.close()


def dml_execute(sql):
    """
    executeを行う(Insert等のDMLで使用する)
    """
    conn = connect()
    cursor = conn.cursor()
    logger = logging.getLogger(__name__)
    try:
        logger.debug("sql={}".format(sql))
        cursor.execute(sql)
        conn.commit()
    except Exception as e:
        conn.rollback()
        logger.exception(message_access.get_message('E900'), e)
        raise e
    finally:
        cursor.close()
        conn.close()

エラー発生時のログ出力をlogger.exceptionに変更