PythonからMySQLに接続する際に使用している接続コードを紹介します。本記事の主な特徴は以下です。
- Dockerによる環境構築およびコンテナ内でのMySQL接続
- Python環境はAnaconda利用
- Pandasデータフレームでのやり取りを意識する
それではやっていきましょう。
▼ Contents
1. Dockerでの環境構築
DokcerおよびDocker-composeでの環境構築を行うため、最初にフォルダー構成を示します。
.
|-- docker-compose.yml
|-- .env
`-- work
|-- Dockerfile
|-- requirements.txt
|-- MySQL_access.py
`-- sample.py
構築用のDocker-compose.ymlの構成は以下です。環境変数は”.env”ファイルにてご指定ください。
version: '3'
volumes:
db_data:
services:
mysql:
image: mysql:8.0
container_name: mysql
environment:
- MYSQL_DATABASE='mysql_db'
- MYSQL_ROOT_USER='root'
- MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
- MYSQL_USER=${MYSQL_USER}
- MYSQL_PASSWORD=${MYSQL_PASSWORD}
- MYSQL_ROOT_Host='%'
ports:
- '3306:3306'
volumes:
- db_data:/var/lib/mysql
tty: true
stdin_open: true
networks:
- app-net
app:
build:
context: ./work
dockerfile: Dockerfile
ports:
- '5550:8888'
container_name: app
volumes:
- '.work:/work'
tty: true
stdin_open: true
depends_on:
- mysql
links:
- mysql
networks:
- app-net
environment:
- DISPLAY=${DISPLAY}
networks:
app-net:
driver: bridge
続いて、Python-appであるwork/Dockerfileの構成は以下です。
FROM ubuntu:latest
RUN apt-get update && apt-get install -y \
sudo \
wget \
vim \
curl \
unzip
WORKDIR /opt
# anacondaのインストール
RUN wget https://repo.continuum.io/archive/Anaconda3-2020.07-Linux-x86_64.sh && \
sh /opt/Anaconda3-2020.07-Linux-x86_64.sh -b -p /opt/anaconda3 && \
rm -f Anaconda3-2020.07-Linux-x86_64.sh
ENV PATH /opt/anaconda3/bin:$PATH
COPY requirements.txt /opt/app/requirements.txt
WORKDIR /opt/app
RUN apt-get update && apt-get install -y \
python3-dev default-libmysqlclient-dev gcc
RUN pip3 install --upgrade -r requirements.txt
COPY . /opt/app
WORKDIR /
CMD ["jupyter", "lab", "--ip=0.0.0.0", "--allow-root", "--LabApp.token=''"]
pipでダウンロードするライブラリを指定するrequirements.txtは以下としています。
pip
pandas
sqlalchemy
mysqlclient
mysql-connector-python
以下のコマンドプロンプトなどで、docker-compose.ymlのあるファオルダーでコマンドでコンテナを作成すれば完了です。
$docker-compose up -d --build
なお、Dockerでの環境構築が不明な方はかめさんのUdemy講座の受講をおすすめします。
2. MySQL接続用のPythonコード例
“mysql_db”に”sample_table”を作成して、MySQLにデータを入力する。その後、入力したデータをPandas DataFrameとして抽出するまで行います。ここでは主に”MySQL_access.py”と”sample.py”を編集していきます。
2-1. ライブラリインポート、環境設定
必要なライブラリとMySQLへ接続するための、変数などを設定します。パスワードなどの変数は、ご自身の環境に合わせてご変更ください。
import pandas as pd
import mysql.connector
from mysql.connector import errorcode
select_table = 'mysql_db'
select_table = 'sample_table'
config = {
"host": "<mysql>",
"port": "<port>" ,
"user": "<username>",
"password": "<password>",
"database": "mysql_db",
}
2-2. 接続テストとMySQL接続解除
MySQLとの接続が行われているか確認するため、クラス変数を読みだした際に、MySQLとの接続を実施するinit関数を定義します。確認したいエラーがあれば、ご自身でelif文を追加すればOKです。また、MySQlとの接続を解除するための関数も作成しておきます。
class access_MySQL():
""" class読込時にMySQLへの接続を確認 """
def __init__(self):
try:
self.con = mysql.connector.connect(**config)
self.cur = self.con.cursor()
except mysql.connector.Error as err:
""" アクセスできない場合 """
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("Something is wrong with your user name or password")
""" データベースがない場合 """
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print("Database does not exist")
else:
print('Unknown error')
def mysql_close(self):
""" MySQLとの接続を切断する """
self.cur.close()
self.con.close()
2-3. MySQLにテーブルを作成する
次に”MySQL_access.py”にはテーブル作成用の関数を用意します。そして、”sample.py”にテーブルを作成するSQLクエリ文を記載して、”MySQL_access.py”を呼び出しテーブルを作成します。
テーブルはサンプルのため、id(AUTO_INCREMENT, PrimaryKey)と、text(Varchar(12)を持つテーブルとします。
def create_table(self, query):
try:
self.cur.execute("DROP TABLE IF EXISTS {}".format(select_table))
self.cur.execute(query)
return True, 'Success to create table'
except mysql.connector.Error as err:
return False, str(err.errno) + " : " + query
from MySQL_access import access_MySQL
select_table = 'sample_table'
sql = """
CREATE TABLE IF NOT EXISTS {} (
id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
text VARCHAR(12) NOT NULL
) CHARSET=utf8mb4
""".format(select_table)
access_MySQL().create_table(sql)
作成できたかはMySQL Workbenchなどでテーブルが作成されているのを確認できます。

2-4. MySQLへのデータ入力
作成したテーブルにデータを入力していきます。”MySQL_access.py”にはデータ入力用の関数を定義します。そして、”sample.py”にテーブルへデータを入力(Insert)するSQLクエリ文を記載して、”MySQL_access.py”を呼び出しテーブルへデータを入力します。
pandasデータフレームを意識しているので、入力するデータをあえて一度データフレームに変換している点にご注意ください。
def insert_data(self, query, df):
data = df.values.tolist()
self.cur.executemany(query, data)
self.con.commit()
select_table = 'sample_table'
sql = ('''
INSERT INTO {}
(text)
VALUES
(%s)
''').format(select_table)
data = pd.DataFrame(pd.Series("sample", name="text"))
access_MySQL().insert_data(sql, data)
こちらもうまく行けば、MySQLへデータが入力されているはずです。

2-5. データ抽出(Select)関数の定義
データ抽出用の関数を作成して、PandasデータフレームとしてMySQLからデータを抽出していきます。
作成したテーブルからデータを抽出します。”MySQL_access.py”にはデータ抽出用の関数を定義します。そして、”sample.py”にテーブルからデータを抽出(Select)するSQLクエリ文を記載して、”MySQL_access.py”を呼び出しテーブルからデータをデータフレームで抽出します。
def mysql_query(self, query):
try:
self.cur.execute(query)
result = self.cur.fetchall()
if not result :
return False, "Record not found or only create table"
else :
return True, result
except mysql.connector.Error as err:
return False, str(err.errno) + " : " + query
def mysql_column(self, select_table):
""" Tableのカラムを取得する """
columns_query = """ DESC {} """.format(select_table)
columns = self.mysql_query(columns_query)
columns_list = [column[0] for column in columns[1]]
self.mysql_close()
return columns_list
def create_df(self, query, select_table):
df_temp = self.mysql_query(query)
df = pd.DataFrame(df_temp[1], columns=self.mysql_column(select_table))
self.mysql_close()
return df
select_table = 'sample_table'
sql = """
SELECT * FROM {}
""".format(select_table)
df = access_MySQL().create_df(sql, select_table)
df
図3のようにデータを抽出できているはずです。

以上、MySQL接続=>テーブル作成=>データ入力=>データ抽出の流れを確認しました。
3. MySQL接続コードとまとめ
本記事で扱った関数をまとめたコードは以下となります。SQlクエリは毎回変更すると思いますが、MySQLとの接続は基本的に同じなので、缶透かしておくとコード数を減らせるのでおすすめです。
ぜひ、ご自身の環境にあったMySQL接続コードを作成してみてください。
import pandas as pd
import mysql.connector
from mysql.connector import errorcode
config = {
"host": "<mysql>",
"port": "<port>" ,
"user": "<username>",
"password": "<password>",
"database": "mysql_db",
}
class access_MySQL():
def __init__(self):
try:
self.con = mysql.connector.connect(**config)
self.cur = self.con.cursor()
except mysql.connector.Error as err:
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("Something is wrong with your user name or password")
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print("Database does not exist")
else:
print('Unknown error')
def mysql_close(self):
self.cur.close()
self.con.close()
def create_table(self, query):
try:
self.cur.execute("DROP TABLE IF EXISTS {}".format(select_table))
self.cur.execute(query)
return True, 'Success to create table'
except mysql.connector.Error as err:
return False, str(err.errno) + " : " + query
def insert_data(self, query, df):
data = df.values.tolist()
self.cur.execute(query, data)
self.con.commit()
def mysql_column(self, select_table):
columns_query = """ DESC {} """.format(select_table)
columns = self.mysql_query(columns_query)
columns_list = [column[0] for column in columns[1]]
self.mysql_close()
return columns_list
def mysql_query(self, query):
try:
self.cur.execute(query)
result = self.cur.fetchall()
if not result :
return False, "Record not found or only create table"
else :
return True, result
except mysql.connector.Error as err:
return False, str(err.errno) + " : " + query
def create_df(self, query, select_table):
df_temp = self.mysql_query(query)
df = pd.DataFrame(df_temp[1], columns=self.mysql_column(select_table))
self.mysql_close()
return df
それでは良いPythonライフを!

他の記事もご欄いただけると幸いです。

Twitterもやってまーす。
ブログを楽しく読ませて頂いております。
Docker-compose.ymlのファイルをコピーしてdocker-compose up -d –buildを実施したのですが、以下のようなエラーが発生しました。よろしければ教えて頂けると幸いです。
ERROR: yaml.scanner.ScannerError: mapping values are not allowed here
in “./docker-compose.yml”, line 9, column 19
yosukeさん
コメントありがとうございます。また、返信遅れてすみません。今週バタバタしてまして。
エラーを見ると、line9の19行目なので、「”」の文字が認識されないようです。
お手数ですが、mysqlの「”」を2つ削除して、もう一度実行いただけますか?
ブログの方も、コードを修正してみました。
うまく実行できると幸いです。