libs.utils.dbutil

  1"""
  2libs/utils/dbutil.py
  3"""
  4
  5import logging
  6import sqlite3
  7from contextlib import closing
  8from importlib.resources import files
  9from typing import TYPE_CHECKING, Any, Optional, Union
 10
 11import libs.global_value as g
 12
 13if TYPE_CHECKING:
 14    from pathlib import Path
 15
 16
 17def connection(database_path: Union["Path", str]) -> sqlite3.Connection:
 18    """
 19    DB接続共通処理
 20
 21    Args:
 22        database_path (Union[Path, str]): データベースファイル
 23
 24    Returns:
 25        sqlite3.Connection: オブジェクト
 26
 27    """
 28    conn = sqlite3.connect(
 29        database=f"file:{database_path}",
 30        # detect_types=sqlite3.PARSE_DECLTYPES,
 31        uri=True,
 32    )
 33    conn.row_factory = sqlite3.Row
 34
 35    return conn
 36
 37
 38def execute(query: str, params: Optional[dict[str, Any]] = None) -> list[dict[str, Any]]:
 39    """
 40    クエリ実行
 41
 42    Args:
 43        query (str): 実行クエリ
 44        params (Optional[dict[str,Any]]): プレースホルダ
 45
 46    Returns:
 47        list[dict[str, Any]]: 実行結果
 48
 49    """
 50    if not params:
 51        params = g.params.placeholder()
 52
 53    ret: list[dict[str, Any]] = []
 54
 55    g.params.update_from_dict(params)
 56    query = g.params.query_modification(query)
 57
 58    if g.args.verbose & 0x01:
 59        print(f">>> params={g.params.placeholder()}")
 60        print(f">>> SQL -> {g.cfg.setting.database_file}\n{g.params.named_query(query)}")
 61
 62    with closing(connection(g.cfg.setting.database_file)) as conn:
 63        try:
 64            rows = conn.execute(query, params)
 65            if conn.total_changes:
 66                conn.commit()
 67        except sqlite3.OperationalError as err:
 68            logging.error("OperationalError: %s", err)
 69            logging.error("params=%s", g.params.placeholder())
 70            logging.error("query: %s", g.params.named_query(query))
 71            return ret
 72
 73        for row in rows.fetchall():
 74            ret.append(dict(row))
 75
 76        if g.args.verbose & 0x02:
 77            print("=" * 80)
 78            print(ret)
 79
 80    return ret
 81
 82
 83def query(keyword: str) -> str:
 84    """
 85    SQLクエリを返す
 86
 87    Args:
 88        keyword (str): SQL選択キーワード
 89
 90    Raises:
 91        ValueError: 未定義のキーワード
 92
 93    Returns:
 94        str: SQL文
 95
 96    """
 97    sql_tables: dict[str, str] = {
 98        # テーブル作成
 99        "CREATE_TABLE_MEMBER": "table/member.sql",
100        "CREATE_TABLE_ALIAS": "table/alias.sql",
101        "CREATE_TABLE_TEAM": "table/team.sql",
102        "CREATE_TABLE_RESULT": "table/result.sql",
103        "CREATE_TABLE_REMARKS": "table/remarks.sql",
104        "CREATE_TABLE_WORDS": "table/words.sql",
105        "CREATE_TABLE_RULE": "table/rule.sql",
106        # VIEW作成
107        "CREATE_VIEW_INDIVIDUAL_RESULTS": "view/individual_results.sql",
108        "CREATE_VIEW_GAME_RESULTS": "view/game_results.sql",
109        "CREATE_VIEW_GAME_INFO": "view/game_info.sql",
110        "CREATE_VIEW_REGULATIONS": "view/regulations.sql",
111        # INDEX作成
112        "CREATE_INDEX": "table/index.sql",
113        # 情報取得
114        "GAME_INFO": "game.info.sql",
115        "RESULTS_INFO": "results.info.sql",
116        "MEMBER_INFO": "member.info.sql",
117        "TEAM_INFO": "team.info.sql",
118        "REMARKS_INFO": "remarks.info.sql",
119        "RECORD_INFO": "record.info.sql",
120        # 集計
121        "SUMMARY_GAMEDATA": "summary/gamedata.sql",
122        "SUMMARY_DETAILS": "summary/details.sql",
123        "SUMMARY_DETAILS2": "summary/details2.sql",
124        "SUMMARY_RESULTS": "summary/results.sql",
125        "SUMMARY_TOTAL": "summary/total.sql",
126        "SUMMARY_VERSUS_MATRIX": "summary/versus_matrix.sql",
127        "RANKING_RESULTS": "ranking/results.sql",
128        "RANKING_RATINGS": "ranking/ratings.sql",
129        "REPORT_PERSONAL_DATA": "report/personal_data.sql",
130        "REPORT_COUNT_DATA": "report/count_data.sql",
131        "REPORT_MONTHLY": "report/monthly.sql",
132        "REPORT_RESULTS_LIST": "report/results_list.sql",
133        "REPORT_WINNER": "report/winner.sql",
134        "REPORT_MATRIX_TABLE": "report/matrix_table.sql",
135        "REPORT_COUNT_MOVING": "report/count_moving.sql",
136        #
137        "RESULT_INSERT": "general/result_insert.sql",
138        "RESULT_UPDATE": "general/result_update.sql",
139        "RESULT_DELETE": "general/result_delete.sql",
140        #
141        "REMARKS_SELECT": "general/remarks_select.sql",
142        "REMARKS_INSERT": "general/remarks_insert.sql",
143        "REMARKS_DELETE_ALL": "general/remarks_delete_all.sql",
144        "REMARKS_DELETE_ONE": "general/remarks_delete_one.sql",
145        "REMARKS_DELETE_COMPAR": "general/remarks_delete_compar.sql",
146        #
147        "WORDS_INSERT": "general/words_insert.sql",
148        #
149        "SELECT_ALL_RESULTS": "general/select_all_results.sql",
150    }
151
152    if query_path := sql_tables.get(keyword):
153        with open(str(files("files.queries").joinpath(query_path)), "r", encoding="utf-8") as queryfile:
154            return str(queryfile.read()).strip()
155    else:
156        raise ValueError(f"Unknown keyword: {keyword}")
157
158
159def table_info(conn: sqlite3.Connection, table_name: str) -> dict[str, Any]:
160    """
161    テーブルのスキーマを取得して辞書で返す
162
163    Args:
164        conn (sqlite3.Connection): オブジェクト
165        table_name (str): テーブル名
166
167    Returns:
168        dict[str, Any]: スキーマ
169
170    """
171    rows = conn.execute(f"pragma table_info('{table_name}');")
172    schema = {row["name"]: dict(row) for row in rows.fetchall()}
173
174    return schema