概要
-
mysql-connector-pythonを使った、PythonからMySQLの情報を取得する方法を紹介します- 前半: MySQLの基本的な操作
- 後半:
mysql-connector-pythonを使ったPythonからMySQLのデータを取得
MySQLのデータベース作成とクエリによる読み込み
参考
- MySQLの基本操作を参考にしました
- その他下記の書籍をときどき参照しました
環境構築
-
MySQL環境構築の忘備録 - ikeh1024のブログ
- （試行錯誤して苦しんだのでまとまってないのは悪しからず…）
NySQLへ接続
- ターミナルからMySQLへ接続します
mysql -u root -p
Enter password:
- データベースを確認します
mysql> SHOW DATABASES;
+--------------------+
| Database |
+--------------------+
| xxx |
| xxxx |
| information_schema |
| mysql |
| performance_schema |
| sys |
+--------------------+
6 rows in set (0.01 sec)
データベースの作成（クエリの実行）
-
demo_databaseという名前でデータベースを作成します
mysql> CREATE database demo_database;
Query OK, 1 row affected (0.04 sec)
mysql> SHOW DATABASES;
+--------------------+
| Database |
+--------------------+
| demo_database |
| xxx |
| xxxx |
| information_schema |
| mysql |
| performance_schema |
| sys |
+--------------------+
7 rows in set (0.00 sec)
- 使用するデータベースを選択します
- 作成したばかりなので勿論テーブルは空ですね
mysql> USE demo_database;
Database changed
mysql> SHOW tables;
Empty set (0.00 sec)
データベースの作成（MySQLダンプファイルの読み込み）
- 参考までにMySQLダンプファイルを読み込んでデータベースを作成する方法を紹介します
- 今回は使用しません
- 適当な名称で
CREATE database "test_db";として空のデータベースを作り、下記のように読み込めばOKです
パスやファイル名を指定して、コマンドライン上からデータベースにdumpファイルをインポートする
-- 取り込みたいデータファイルと同じ場所まで移動して実行 $ mysql -u [MySQLのユーザ名] -p test_db < example.sql
-
.sqlのスクリプトを適用したい場合は下記を参照
mysql -u root -p 「DB名」 < 「対象ファイル」
テーブルの作成
- テーブルの作成と確認
/* CREATE TABLE */
CREATE TABLE sample_table
(
sample_id int(11) NOT NULL COMMENT 'サンプルID',
sample_name varchar(50) NOT NULL COMMENT 'サンプル名',
UNIQUE(sample_id, sample_name)
)
COMMENT='テーブル用のコメント'
;
mysql> SHOW tables;
+-------------------------+
| Tables_in_demo_database |
+-------------------------+
| sample_table |
+-------------------------+
1 row in set (0.02 sec)
テーブルへデータを登録
- 適当にデータを追加します
/* Adding Data */
SET @sample_id_first = 1;
SET @sample_id_second = 2;
SET @sample_id_third = 3;
-- Comment
INSERT INTO sample_table VALUES (@sample_id_first, "1番目");
INSERT INTO sample_table VALUES (@sample_id_second, "2番目");
INSERT INTO sample_table VALUES (@sample_id_third, "3番目");
データの確認: クエリ
- 先程のデータが登録されているか確認します
- また下記のクエリを、後述の
mysql-connector-pythonを使ってPythonから呼び出します
SELECT
sample_table.sample_id AS id,
sample_table.sample_name AS name
FROM sample_table
/* ソート設定 */
ORDER BY sample_table.sample_id
;
+----+---------+
| id | name |
+----+---------+
| 1 | 1番目 |
| 2 | 2番目 |
| 3 | 3番目 |
+----+---------+
3 rows in set (0.00 sec)
データの確認: SQLクライアントツール
- DBeaver CommunityというSQLクライアントツールを使って、GUI的にデータベースを閲覧してみます
- GUIで楽に確認できるのが良いですね
- 接続設定は以下の通りです
- またPCを再起動してアプリを起動した場合、DBの接続がエラーになるので、一旦ターミナル経由でログインが必要でした
mysql -u root -p
mysql-connector-pythonを使用してMySQLの情報を取得
参考
-
MySQL Connector/Python Developer Guide
- 公式サイト
-
10.6.4 cursor.MySQLCursorDict Class
- データベースから取得した結果を辞書型で取得する
- データベースから取得した結果を辞書型で取得する
mysql-connector-pythonのインストール
- 下記の通りインストールします
pip install mysql-connector-python
mysql-connector-pythonの使用例
- 先述の同じクエリを、Pythonから実行してMySQLから情報を取得してみます
import mysql.connector
# DBへ接続
conn = mysql.connector.connect(
user='root',
password='password',
host='localhost',
database='demo_database'
)
# DBの接続確認
if not conn.is_connected():
raise Exception("MySQLサーバへの接続に失敗しました")
cur = conn.cursor(dictionary=True) # 取得結果を辞書型で扱う設定
query__for_fetching = """
SELECT
sample_table.sample_id AS id,
sample_table.sample_name AS name
FROM sample_table
ORDER BY sample_table.sample_id
;
"""
cur.execute(query__for_fetching)
for fetched_line in cur.fetchall():
id = fetched_line['id']
name = fetched_line['name']
print(f'{id}: {name}')
- 実行結果は以下の通りです
- 無事SQLからデータを読み込むことができてますね。
1: 1番目
2: 2番目
3: 3番目
- また私は扱いやすくするため、下記の
MysqlConnectorManagerのようなクラスを作成して使用しています。
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
import mysql.connector
@dataclass
class MysqlConnectorManager():
user: str = ""
password: str = ""
host: str = ""
database_name: str = ""
cur: Optional[Any] = None
conn: Optional[Any] = None
@classmethod
def create(cls,
user: str,
password: str,
host: str,
database_name: str,
) -> 'MysqlConnectorManager':
return MysqlConnectorManager(user=user, password=password, host=host, database_name=database_name)
def __del__(self):
self._close_connection()
def start_connection(self) -> None:
"""DBに接続する
"""
self.conn = mysql.connector.connect(
user=self.user,
password=self.password,
host=self.host,
database=self.database_name
)
# DB接続の確認
if not self.conn.is_connected():
raise Exception("MySQLサーバへの接続に失敗しました")
self.cur = self.conn.cursor(dictionary=True)
def _close_connection(self) -> None:
"""DBの接続を切る
"""
self.cur.close
self.conn.close
self.cur = None
self.conn = None
def fetch_contents(self, query: str) -> List[Dict[str, Any]]:
"""DBから情報を取得する
"""
self.cur.execute(query)
fetched_contents: List[Dict[str, Any]] = self.cur.fetchall()
return fetched_contents