SQLAlchemy使ってみた①

概要

既存のデータベースから必要なデータを取得することを目標とする(そのためトランザクションなどは扱わない)。 今回は手始めに環境構築+SQLAlchemyを使ってSakillaデータベースのメタデータ取得を行う。

環境

準備

まず必要なパッケージのインストール。 pythonプログラムからmysqlに接続するにはドライバのインストールも必要なので、mysql公式ドライバのmysql-connector-pythonをインストール。

wd $ pip install  SQLAlchemy=="1.3.0", mysql-connector-python="8.0.31"

またデータベースのサンプルとしてSakila Sample Databaseを使っていく。 直接リンクを叩くなりwget/curlを使うなりでとりあえずデータベースのダウンロード。 zipを展開すると中身こんな感じ。

wd $ ls sakila-db 
sakila-data.sql    sakila-schema.sql  sakila.mwb

Mysqlのローカルサーバーを立ち上げる。

wd $ mysql.server start

ユーザ名、パスワードなど確認(テスト用なのでrootで勘弁してくれ)

wd $ mysql -u root -p
Enter password: ?
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 12
Server version: 8.0.31 Homebrew

Copyright (c) 2000, 2022, Oracle and/or its affiliates.

Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.

Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
mysql>

まずsqlスクリプトを読み込み、sakilaデータベースを作成する。

mysql> source sakila-db/sakila-schema.sql;
mysql> source sakila-db/sakila-data.sql;

データベースが正しくインストールされていることの確認。

mysql> USE sakila;
Database changed
mysql> SHOW FULL TABLES;
+----------------------------+------------+
| Tables_in_sakila           | Table_type |
+----------------------------+------------+
.....

これでMySQLサーバーの準備は完了。 ちなみにサーバーを終了するには、セッションから抜け、プロセスをkillすればよい。

mysql> quit 
wd $ mysql.server stop

SQLAlchemyでDB接続の確認

ユーザー側が触れるAPIは、CoreとORMがあるが、今回はCoreを使っていく。 Coreによるデータベースへの接続方法は単にcreate_engineを呼び出すだけである。ドキュメントに従いurlを生成し、create_engineに渡せばよい。

MetaDataにengineをバインドすることで、sakilaデータベースのメタデータが取得可能になる。 まずモジュールのインポート

import urllib.parse
from typing import List, Optional

from sqlalchemy import create_engine, MetaData
import toml

MySQLサーバーアクセスのための諸々の設定(dialect, driver, username, password, host)はとりあえずtomlファイルに書いておき読み込む。 craete_urlメソットでurlを生成。

class SqlSetting:
    """Read sql setting from toml file"""
    def __init__(self, dialect:str, setting_file: str):
        parsed_toml = toml.load(setting_file)
        self.__dialect = dialect
        self.__username = parsed_toml[dialect]['username']
        self.__password = urllib.parse.quote_plus(parsed_toml[dialect]['password']) 
        self.__host = parsed_toml[dialect]['host']
        self.__driver = parsed_toml[dialect]['driver']
        self.__database = parsed_toml[dialect]['database']

    def create_url(self, dbname: str): 
        return f'{self.__dialect}+{self.__driver}://{self.__username}:{self.__password}@{self.__host}/{dbname}'

create_eingineにurlを渡し、ローカルサーバーのSakilaデータベースにアクセス。 echo=Trueを指定すると、ログが出力され、発行されるsql文を確認することができる。

    setting = SqlSetting(dialect='mysql', setting_file="setting.toml")
    engine=create_engine(setting.create_url(dbname='sakila'), echo=True)

データベースのメタデータを取得するにはMetaDataをインスタンス化し、engineとバインドさせれば良い。

    # bind engine to metadata
    metadata = MetaData(bind=engine)  
    metadata.reflect(only=["actor"]) # onlyを指定すると特定のテーブルのみクエリ(mysqlならshow create table)が発行される
    # show tables 
    for table in metadata.sorted_tables:
        print("table name: ", table.name)
        print("table columns: ", table.columns)
    #>> table name: actor
    #>> table columns:  ['actor.actor_id', 'actor.first_name', 'actor.last_name', 'actor.last_update']
    #>> ...
    #>>table name store
    #>>able columns ['store.store_id', 'store.manager_staff_id', 'store.address_id', 'store.last_update']

例としてactorテーブルのメタデータを取得してみる。

    actor_table = metadata.tables["actor"] 
    print(actor_table.columns)
    #>> ['actor.actor_id', 'actor.first_name', 'actor.last_name', 'actor.last_update']

カラムには下記方法でもアクセス可能。

    print(actor_table.c.actor_id)
    #>> actor.actor_id

次はselect文使ってデータ取得してみる。