MySQLデータをPythonを使ってエクスポートする方法

MySQLデータをCSVでエクスポートする方法を紹介します。
対応はDockerなどの仮想環境、ssh接続+サーバー環境、PlanetScaleです。

Dockerなどの仮想環境DB / PlanetScale

仮想環境.ver

import pandas as pd
import pymysql.cursors

class ConnectLocal:

    def result(self,sql,dir_path):
        connection = pymysql.connect(
            host='127.0.0.1',
            user='hoge_user',
            password='hoge_pwd',
            database='hoge_db',
            cursorclass=pymysql.cursors.DictCursor,
            port=13306 (自身で設定したポート)
        )

        exam_sql = "select * from users where id = 1"
        dir_path = "exports"

        df = pd.read_sql(sql=exam_sql, con=connection)
        df.to_csv(f'{dir_path}/local_data.csv')
        connection.close()

        return "successfully connect local!"

PlanetScale.ver

import pymysql.cursors
import pandas as pd

class ConnectPlanet:

    def result(self,sql,dir_path):
        
        connection = pymysql.connect(
            host='hoge_host',
            user='hoge_user',
            password='hoge_pwd',
            database='hoge_db',
            cursorclass=pymysql.cursors.DictCursor,
            ssl={'ca': '/etc/ssl/cert.pem', 'check_hostname': False})

        exam_sql = "select * from users where id = 1"
        dir_path = "exports"

        df = pd.read_sql(sql=exam_sql, con=connection)
        df.to_csv(f'{dir_path}/planet_data.csv')
        connection.close()

        return "successfully connect planet!"

Dockerの場合、自身で設定したDBをまんまコネクターに値を入れるだけです。
コードの中のhoge_とはfuga_系が設定値になります。
PlanetScaleの場合、sslで接続することが要求されるので、自身のcert.pemのパスを設定して接続します。

SQL操作+エクスポート

pandasのread_sql()で、第１引数に任意のSQL、第二引数にconnectionを指定すると該当データをDataFrameで返してくれます。あとはto_csv()で任意の場所に落としてあげるとデータを取得できます。
階層を掘りたいときは上記のように書きますが、ディレクトリがないとエラーが出るので、

your_dir = "exports"
os.mkdir(your_dir) if not os.path.isdir(your_dir) else None

みたいな感じでto_csv()前に置くといいと思います。

コネクターがない人は下記でインストールしてください。

$ pip3 install pymysql

pandasない人は
$ pip3 install pandas

ssh接続+サーバー環境

from sshtunnel import SSHTunnelForwarder
import pandas as pd
import pymysql.cursors

class ConnectPub:

    def result(self,sql,dir_path):
        sshOptions = {
            "host": "hoge_host",
            "ssh_username": "hoge_username",
            "ssh_password": "hoge_psw"
        }

        with SSHTunnelForwarder(
            (sshOptions["host"], 22),
            ssh_username=sshOptions["ssh_username"],
            ssh_password=sshOptions["ssh_password"],
            remote_bind_address=("localhost", 3306)
        ) as server:
            fuga_connection = pymysql.connect(host='localhost',
                            port=server.local_bind_port,
                            user="fuga_user",
                            passwd="fuga_pwd",
                            db="fuga_db",
                            charset='utf8',
                            cursorclass=pymysql.cursors.DictCursor,
                    )
            exam_sql = "select * from users where id = 1"
            dir_path = "exports"
            df = pd.read_sql(sql=exam_sql, con=fuga_connection)
            df.to_csv(f'{dir_path}/fuga_data.csv')
            pub_connection.close()
            return "successfully connect ssh and server db!"

仮想環境DBと接続が異なるのはsshでトンネリングして接続する点です。
mysqlに接続する前にSSHTunnelForwarderでbind portを取得します。
そのあとは同じくDBに接続して、データを落とす形になります。

SSHTunnelForwarderがない人は下記でインストールしてください。

$ pip3 install sshtunnel

まとめ

ツールを使ってGUIでポチるのが毎回面倒だと感じる人はこの方法を参考にしてみてください。