{"id":3550,"date":"2025-04-14T21:41:08","date_gmt":"2025-04-14T12:41:08","guid":{"rendered":"https:\/\/www.dogrow.net\/python\/?p=3550"},"modified":"2025-04-14T23:29:42","modified_gmt":"2025-04-14T14:29:42","slug":"blog127-mysql-db%e3%83%89%e3%83%a9%e3%82%a4%e3%83%90%e3%81%ae%e3%83%a9%e3%83%83%e3%83%91%e3%82%af%e3%83%a9%e3%82%b9%e3%82%92%e4%bd%9c%e3%82%8b%e3%80%82","status":"publish","type":"post","link":"https:\/\/www.dogrow.net\/python\/blog127-mysql-db%e3%83%89%e3%83%a9%e3%82%a4%e3%83%90%e3%81%ae%e3%83%a9%e3%83%83%e3%83%91%e3%82%af%e3%83%a9%e3%82%b9%e3%82%92%e4%bd%9c%e3%82%8b%e3%80%82\/","title":{"rendered":"(127) MySQL DB\u30c9\u30e9\u30a4\u30d0\u306e\u30e9\u30c3\u30d1\u30fc\u30af\u30e9\u30b9\u3092\u4f5c\u308b\u3002"},"content":{"rendered":"

1. \u3084\u308a\u305f\u3044\u3053\u3068<\/h1>\n

Python\u30d7\u30ed\u30b0\u30e9\u30e0\u3067 MySQL DB\u306b\u30a2\u30af\u30bb\u30b9\u3059\u308b\u3068\u304d\u3001\u3053\u3053\u6570\u5e74\u306f SQLAlchemy<\/a> \u3092\u4f7f\u3063\u3066\u3044\u308b\u3002
\n\u4eca\u56de\u3001\u305f\u307e\u305f\u307e mysql-connector-python<\/a> \u3092\u4f7f\u3046\u3053\u3068\u306b\u306a\u3063\u305f\u306e\u3067\u3001\u3053\u308c\u3092\u4f7f\u3046\u4fbf\u5229\u30af\u30e9\u30b9\u3067\u30e9\u30c3\u30d7\u3057\u305f\u3044\u3002<\/p>\n

2. \u3084\u3063\u3066\u307f\u308b<\/h1>\n

\u500b\u4eba\u7684\u306a\u5099\u5fd8\u9332\u306a\u306e\u3067\u3001\u89e3\u8aac\u7121\u3057\u3067\u7d50\u679c\u3060\u3051\u3092\u8a18\u9332\u3057\u3066\u304a\u304f\u3002
\n\u203b\u4eca\u5f8c\u306e\u6539\u826f\u6642\u306b\u968f\u6642\u66f4\u65b0\u3057\u3088\u3046\u3002<\/span><\/p>\n

(1\/5) logger_config.py<\/h3>\n
\r\nimport logging\r\n\r\nlogger = logging.getLogger('MyApp')\r\nlogger.setLevel(logging.INFO)\r\n\r\nhandler   = logging.FileHandler('myapp.log', mode='a', encoding='utf-8')\r\nformatter = logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s')\r\nhandler.setFormatter(formatter)\r\nlogger.addHandler(handler)\r\n<\/pre>\n
(2\/5) DBSession.py<\/h3>\n
DB\u63a5\u7d9a\u7ba1\u7406\u30af\u30e9\u30b9<\/p>\n
\r\nfrom logger_config import logger\r\nimport mysql.connector\r\nfrom mysql.connector import Error\r\n\r\nclass DBSession:\r\n    def __init__(self, host, user, password, database):\r\n        try:\r\n            self.conn = mysql.connector.connect(\r\n                host=host,\r\n                user=user,\r\n                password=password,\r\n                database=database\r\n            )\r\n            self._cursor = self.conn.cursor(dictionary=True)\r\n            logger.info("DBSession \u63a5\u7d9a\u6210\u529f")\r\n        except Error as e:\r\n            logger.error(f"DBSession \u63a5\u7d9a\u30a8\u30e9\u30fc: {e}")\r\n            self.conn = None\r\n            self._cursor = None\r\n            raise\r\n\r\n    @property\r\n    def cursor(self):\r\n        if not self.conn or not self._cursor:\r\n            raise RuntimeError("DB\u63a5\u7d9a\u307e\u305f\u306f\u30ab\u30fc\u30bd\u30eb\u304c\u7121\u52b9\u3067\u3059")\r\n        return self._cursor\r\n\r\n    def commit(self):\r\n        if self.conn:\r\n            self.conn.commit()\r\n\r\n    def rollback(self):\r\n        if self.conn:\r\n            self.conn.rollback()\r\n\r\n    def close(self):\r\n        if self._cursor:\r\n            self._cursor.close()\r\n        if self.conn:\r\n            self.conn.close()\r\n        logger.info("DBSession \u5207\u65ad\u5b8c\u4e86")\r\n<\/pre>\n
(3\/5) DBTableBase.py<\/h3>\n
\u5404\u30c6\u30fc\u30d6\u30eb\u7528DAO\u306e\u89aa\u30af\u30e9\u30b9<\/p>\n
\r\nfrom logger_config import logger\r\nfrom typing import List, Dict, Any\r\nimport re\r\n\r\nclass DBTableBase:\r\n    def __init__(self, session, table_name: str, id_column: str):\r\n        if not re.match(r'^\\w+$', table_name):\r\n            raise ValueError("Invalid table name")\r\n        if not re.match(r'^\\w+$', id_column):\r\n            raise ValueError("Invalid id column")\r\n        self.session = session\r\n        self._table_name = table_name\r\n        self._id_column = id_column\r\n        self._ary_value = []\r\n\r\n    @property\r\n    def table_name(self):\r\n        return self._table_name\r\n\r\n    @property\r\n    def id_column(self):\r\n        return self._id_column\r\n\r\n    @property\r\n    def cursor(self):\r\n        if not self.session.conn:\r\n            raise RuntimeError("DB\u63a5\u7d9a\u304c\u7121\u52b9\u3067\u3059")\r\n        return self.session.cursor\r\n\r\n    @property\r\n    def fetchedData(self):\r\n        return self._ary_value\r\n\r\n    def execute(self, sql: str, params: tuple = ()): \r\n        self.cursor.execute(sql, params)\r\n\r\n    def insert(self, columns: List[str], values: List[Any]):\r\n        try:\r\n            placeholders = ", ".join(["%s"] * len(values))\r\n            column_names = ", ".join(columns)\r\n            sql = f"INSERT INTO {self.table_name} ({column_names}) VALUES ({placeholders})"\r\n            self.execute(sql, tuple(values))\r\n        except Exception as e:\r\n            logger.error(f"[{self.__class__.__name__}] INSERT\u30a8\u30e9\u30fc: {e}")\r\n            raise\r\n\r\n    def insert_dict(self, data: Dict[str, Any]):\r\n        self.insert(list(data.keys()), list(data.values()))\r\n\r\n    def select_by_id(self, id_value: Any) -> Dict[str, Any]:\r\n        try:\r\n            sql = f"SELECT * FROM {self.table_name} WHERE {self.id_column} = %s"\r\n            self.execute(sql, (id_value,))\r\n            self._ary_value = [self.cursor.fetchone()]\r\n            return self._ary_value[0]\r\n        except Exception as e:\r\n            logger.error(f"[{self.__class__.__name__}] SELECT\u30a8\u30e9\u30fc: {e}")\r\n            raise\r\n\r\n    def select_by_conditions(self, conditions: Dict[str, Any]) -> List[Dict[str, Any]]:\r\n        try:\r\n            where_clause = " AND ".join([f"{k} = %s" for k in conditions.keys()])\r\n            sql = f"SELECT * FROM {self.table_name} WHERE {where_clause}"\r\n            self.execute(sql, tuple(conditions.values()))\r\n            self._ary_value = self.cursor.fetchall()\r\n            return self._ary_value\r\n        except Exception as e:\r\n            logger.error(f"[{self.__class__.__name__}] \u6761\u4ef6\u691c\u7d22\u30a8\u30e9\u30fc: {e}")\r\n            raise\r\n\r\n    def select_all(self) -> List[Dict[str, Any]]:\r\n        try:\r\n            sql = f"SELECT * FROM {self.table_name} ORDER BY {self.id_column}"\r\n            self.execute(sql)\r\n            self._ary_value = self.cursor.fetchall()\r\n            return self._ary_value\r\n        except Exception as e:\r\n            logger.error(f"[{self.__class__.__name__}] \u5168\u4ef6\u53d6\u5f97\u30a8\u30e9\u30fc: {e}")\r\n            raise\r\n\r\n    def update(self, updates: Dict[str, Any], conditions: Dict[str, Any]):\r\n        try:\r\n            set_clause = ", ".join([f"{k} = %s" for k in updates.keys()])\r\n            where_clause = " AND ".join([f"{k} = %s" for k in conditions.keys()])\r\n            sql = f"UPDATE {self.table_name} SET {set_clause} WHERE {where_clause}"\r\n            values = list(updates.values()) + list(conditions.values())\r\n            self.execute(sql, tuple(values))\r\n        except Exception as e:\r\n            logger.error(f"[{self.__class__.__name__}] UPDATE\u30a8\u30e9\u30fc: {e}")\r\n            raise\r\n\r\n    def update_by_id(self, id_value: Any, update_column: str, new_value: Any):\r\n        self.update({update_column: new_value}, {self.id_column: id_value})\r\n\r\n    def delete_by_id(self, id_value: Any):\r\n        try:\r\n            sql = f"DELETE FROM {self.table_name} WHERE {self.id_column} = %s"\r\n            self.execute(sql, (id_value,))\r\n        except Exception as e:\r\n            logger.error(f"[{self.__class__.__name__}] DELETE\u30a8\u30e9\u30fc: {e}")\r\n            raise\r\n<\/pre>\n
(4\/5) PriceTable.py<\/h3>\n
\u500b\u5225\u306e\u30c6\u30fc\u30d6\u30eb\u7528DAO\u30af\u30e9\u30b9
\n\u203b\u5fc5\u305aDBTableBase\u3092\u7d99\u627f\u3059\u308b\u3053\u3068<\/span><\/p>\n
\r\nfrom DBTableBase import DBTableBase\r\nfrom datetime import date\r\nfrom typing import Union\r\n\r\nclass PriceTable(DBTableBase):\r\n    def __init__(self, session):\r\n        super().__init__(session, 'tbl_price', 'p_id')\r\n\r\n    def insert_price(self, pd_id: int, p_date: date, p_price: Union[int, float], s_id: int):\r\n        self.insert_dict({\r\n            'pd_id':   pd_id,\r\n            'p_date':  p_date,\r\n            'p_price': p_price,\r\n            's_id':    s_id,\r\n        })\r\n\r\n    def update_price(self, p_id: int, new_price: Union[int, float]):\r\n        self.update_by_id(p_id, 'p_price', new_price)\r\n<\/pre>\n
(5\/5) _usage.py<\/h3>\n
\u4f7f\u7528\u4f8b<\/p>\n
\r\nfrom datetime import date\r\nfrom DBSession import DBSession\r\nfrom ShopTable import ShopTable\r\nfrom ProductTable import ProductTable\r\n\r\n# DB\u63a5\u7d9a\r\nsession = DBSession('localhost', 'xxxxx', 'yyyyy', 'my_db')\r\nshop    = ShopTable(session)\r\nproduct = ProductTable(session)\r\n\r\n# (1) shop\u30c6\u30fc\u30d6\u30eb\u306e\u30ec\u30b3\u30fc\u30c9\u3092\u8868\u793a\u3059\u308b\u3002\r\nary_res1 = shop.select_all()\r\nprint(ary_res1)\r\nary_res2 = shop.fetchedData\r\nprint(ary_res2)\r\n\r\n# (2) product\u30c6\u30fc\u30d6\u30eb\u306b\u30ec\u30b3\u30fc\u30c9\u3092\u8ffd\u52a0\u3059\u308b\u3002\r\nproduct.insert_product('Red Apple', '1')\r\nsession.commit()\r\n<\/pre>\n
3. MariaDB\u516c\u5f0f\u30e9\u30a4\u30d6\u30e9\u30ea\u3092\u4f7f\u3046\u5834\u5408<\/h1>\n
MariaDB\u7528\u306b\u6700\u9069\u5316\u3055\u308c\u3066\u3044\u308b\u306e\u3067\u3001MariaDB\u3092\u4f7f\u3046\u5834\u5408\u306f\u3053\u3061\u3089\u306e\u65b9\u304c\u826f\u3044\u306f\u305a\u3002<\/p>\n
\u5143\u304c\u540c\u3058\u30bd\u30d5\u30c8\u30a6\u30a7\u30a2\u306a\u306e\u3067\u5927\u304d\u306a\u5dee\u306f\u306a\u3044\u3002
\nDBSession.py\u306e import\u3060\u3051\u5909\u3048\u308c\u3070\u52d5\u304f\u3002\uff082025.04.14\u6642\u70b9\uff09<\/p>\n
\r\nfrom logger_config import logger\r\nimport mariadb\r\nfrom mariadb import Error\r\n\r\nclass DBSession:\r\n    def __init__(self, host, user, password, database):\r\n        try:\r\n            self.conn = mariadb.connect(\r\n                host=host,\r\n                user=user,\r\n                password=password,\r\n                database=database\r\n            )\r\n            self._cursor = self.conn.cursor(dictionary=True)\r\n            logger.info("DBSession \u63a5\u7d9a\u6210\u529f")\r\n        except Error as e:\r\n            logger.error(f"DBSession \u63a5\u7d9a\u30a8\u30e9\u30fc: {e}")\r\n            self.conn = None\r\n            self._cursor = None\r\n            raise\r\n\r\n    @property\r\n    def cursor(self):\r\n        if not self.conn or not self._cursor:\r\n            raise RuntimeError("DB\u63a5\u7d9a\u307e\u305f\u306f\u30ab\u30fc\u30bd\u30eb\u304c\u7121\u52b9\u3067\u3059")\r\n        return self._cursor\r\n\r\n    def commit(self):\r\n        if self.conn:\r\n            self.conn.commit()\r\n\r\n    def rollback(self):\r\n        if self.conn:\r\n            self.conn.rollback()\r\n\r\n    def close(self):\r\n        if self._cursor:\r\n            self._cursor.close()\r\n        if self.conn:\r\n            self.conn.close()\r\n        logger.info("DBSession \u5207\u65ad\u5b8c\u4e86")\r\n<\/pre>\n
\n","protected":false},"excerpt":{"rendered":"
1. \u3084\u308a\u305f\u3044\u3053\u3068 Python\u30d7\u30ed\u30b0\u30e9\u30e0\u3067 MySQL DB\u306b\u30a2\u30af\u30bb\u30b9\u3059\u308b\u3068\u304d\u3001\u3053\u3053\u6570\u5e74\u306f SQLAlchemy \u3092\u4f7f\u3063\u3066\u3044\u308b\u3002 \u4eca\u56de\u3001\u305f\u307e\u305f\u307e mysql-connector-python \u3092\u4f7f\u3046\u3053\u3068\u306b\u306a\u3063\u305f\u306e\u3067\u3001\u3053\u2026 \u7d9a\u304d\u3092\u8aad\u3080 »<\/a><\/span><\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[63],"tags":[],"class_list":["post-3550","post","type-post","status-publish","format-standard","hentry","category-database"],"views":658,"amp_enabled":true,"_links":{"self":[{"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/posts\/3550","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/comments?post=3550"}],"version-history":[{"count":13,"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/posts\/3550\/revisions"}],"predecessor-version":[{"id":3563,"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/posts\/3550\/revisions\/3563"}],"wp:attachment":[{"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/media?parent=3550"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/categories?post=3550"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.dogrow.net\/python\/wp-json\/wp\/v2\/tags?post=3550"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}