【Docker, Python】MySQL connectorサンプルコード

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講座の受講をおすすめします。

関連記事データサイエンス向けのおすすめUdemy講座【結論:かめさん講座】

2021.12.27

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ライフを！