chatgpt に作ってもらった python の mysql 操作クラス、動作未確認

DbAccess.py

from pymysql import connect
from pymysql.cursors import DictCursor
from typing import Optional, List, Dict, Any
from datetime import datetime
from .query_builder import QueryBuilder  # QueryBuilder は PHP のものに相当する関数が必要

class DbAccess:
    _instance: Optional["DbAccess"] = None

    def __init__(self):
        self.pdo = connect(
            host=os.getenv("DB_HOST"),
            user=os.getenv("DB_USERNAME"),
            password=os.getenv("DB_PASSWORD"),
            db=os.getenv("DB_DATABASE"),
            charset='utf8mb4',
            cursorclass=DictCursor,
            autocommit=False
        )

    @classmethod
    def singleton(cls) -> "DbAccess":
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance

    def select(
        self,
        table: str,
        where: Dict[str, Any] = {},
        order: Dict[str, str] = {},
        limit: Optional[int] = None,
        offset: Optional[int] = None,
    ) -> List[Dict[str, Any]]:
        where_clause, params = QueryBuilder.build_where(where)
        order_clause = QueryBuilder.build_order_by(order)

        sql = f"SELECT * FROM {table} {where_clause} {order_clause}"
        if limit is not None:
            sql += f" LIMIT {limit}"
        if offset is not None:
            sql += f" OFFSET {offset}"

        with self.pdo.cursor() as cursor:
            cursor.execute(sql, params)
            return cursor.fetchall()

    def select_one(
        self,
        table: str,
        where: Dict[str, Any] = {},
        order: Dict[str, str] = {}
    ) -> Optional[Dict[str, Any]]:
        results = self.select(table, where, order, limit=1)
        return results[0] if results else None

    def insert(self, table: str, data: Dict[str, Any]) -> bool:
        data["created_at"] = datetime.now()
        data["updated_at"] = datetime.now()

        columns = ', '.join(data.keys())
        placeholders = ', '.join(['%s'] * len(data))
        values = list(data.values())

        sql = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
        with self.pdo.cursor() as cursor:
            cursor.execute(sql, values)
        return True

    def update(self, table: str, update_data: Dict[str, Any], where: Dict[str, Any]) -> bool:
        assert update_data and where

        update_data["updated_at"] = datetime.now()
        set_clause = ', '.join([f"{k} = %s" for k in update_data])
        values = list(update_data.values())

        where_clause, where_values = QueryBuilder.build_where(where)
        sql = f"UPDATE {table} SET {set_clause} {where_clause}"

        with self.pdo.cursor() as cursor:
            cursor.execute(sql, values + where_values)
        return True

    def delete(self, table: str, where: Dict[str, Any]) -> bool:
        where_clause, params = QueryBuilder.build_where(where)
        sql = f"DELETE FROM {table} {where_clause}"

        with self.pdo.cursor() as cursor:
            cursor.execute(sql, params)
        return True

    def bulk_insert(self, table: str, rows: List[Dict[str, Any]]) -> bool:
        if not rows:
            raise Exception("挿入するデータがありません")

        columns = list(rows[0].keys()) + ['created_at', 'updated_at']
        placeholders = '(' + ', '.join(['%s'] * len(columns)) + ')'
        values = []

        for row in rows:
            row_values = list(row.values())
            row_values.append(datetime.now())
            row_values.append(datetime.now())
            values.extend(row_values)

        all_placeholders = ', '.join([placeholders] * len(rows))
        sql = f"INSERT INTO {table} ({', '.join(columns)}) VALUES {all_placeholders}"

        with self.pdo.cursor() as cursor:
            cursor.execute(sql, values)
        return True

QueryBuilder.py

from typing import Dict, List, Tuple, Union, Any

class QueryBuilder:

    @staticmethod
    def build_where(where: Dict[str, Union[str, int, List[Union[str, int]], None]]) -> Tuple[str, List[Any]]:
        params: List[Any] = []
        where_columns: List[str] = []

        for key, value in where.items():
            if value is None:
                where_columns.append(f"{key} IS NULL")
            elif isinstance(value, list):
                if "BETWEEN" in key:
                    where_columns.append(f"{key} %s AND %s")
                    params.extend(value)
                else:
                    if len(value) > 0:
                        placeholders = ', '.join(['%s'] * len(value))
                        where_columns.append(f"{key} IN ({placeholders})")
                        params.extend(value)
                    else:
                        where_columns.append("1 = 0")
            else:
                if " " in key:
                    where_columns.append(f"{key} %s")
                else:
                    where_columns.append(f"{key} = %s")
                params.append(value)

        where_clause = ''
        if where_columns:
            where_clause = 'WHERE ' + ' AND '.join(where_columns)

        return where_clause, params

    @staticmethod
    def build_order_by(order: Union[Dict[str, str], List[str]]) -> str:
        if not order:
            return ''

        order_columns: List[str] = []

        if isinstance(order, dict):
            for column, sort in order.items():
                order_columns.append(f"{column} {sort}")
        elif isinstance(order, list):
            order_columns = order

        return ' ORDER BY ' + ', '.join(order_columns)