DataClassでクラス定義とアノテーションからSQL文を作りたくて試してみた。

ブツ

リポジトリ

NAME='Python.DataClass.SqlBuilder.20221013125810'
git clone https://github.com/ytyaru/$NAME
cd $NAME/src
python run.py

Pythonの勉強をする

今回はDataClassを使ってPythonの勉強をしてみる。題材はORMライブラリ。複雑なので絶対に頓挫するのは目に見えている。でもやってみたい。

DataClass

Python3.7以降にはDataClassという機能がある。これはC言語でいう構造体のこと。classに@dataclassデコレータを付与することで実装できる。メンバ変数に型アノテーションを付与するのが特徴。

from dataclasses import dataclass

@dataclass
class Human:
    name: str
    age: int = 0

ORM

ORMはオブジェクト関係マッピングのこと。オブジェクト指向言語のクラスとDBのリレーションシップ・テーブルを相互変換すること。大抵はORMライブラリがある。

たとえば以下のようなテーブルがある。

create table Human(name text, age integer);

以下のようなレコードがあったとする。

insert into Human values('Yamada', 0);
insert into Human values('Suzuki', 123);

name	age
`Yamada`	`0`
`Suzuki`	`123`

これを以下のようなPythonのクラスで取得したい。

class Human:
    name: str
    age: int

とりあえず適当にORM.Human.select(name='Yamada')というAPIでレコード取得できると想定する。

yamada = ORM.Human.select(name='Yamada')

レコードはHumanクラスになる。nameやageプロパティにアクセスすることでレコードのフィールド値を参照できる。

yamada.name # 'Yamada'
yamada.age  # 0

おなじく挿入・削除、テーブル作成APIも用意する。それぞれHumanクラスのインスタンスを渡すことで操作できる。細かい点はそのとき考える。

ORM.select(Human(name='Yamada'))
ORM.insert(Human(name='Yamada', age=0))
ORM.update(Human(name='Yamada', age=0))
ORM.create_table(Human)

もしORMを使わなければsqlite3ライブラリを使って以下のようにSQL文をもちいて取得することになる。結果は２次元配列。SQL文を書かねばならない上に名前でアクセスできないため可読性が低い。

import sqlite3
conn = sqlite3.connect('example.db')
c = conn.cursor()
c.execute('create table human(name text, age integer)')
c.execute("insert into human values('Yamada', 0)")
c.execute("insert into human values('Suzuki', 123)")
c.execute("select * from human where name='Yamada'")
yamada = c.fetchone()
assert yamada[0] == 'Yamada'
assert yamada[1] == 0

SQLAlchemy

SQLAlchemyはPython用ORMライブラリ。ちゃんとやりたいならこれを使ったほうがいい。

ソースコード

run.py

#!/usr/bin/env python3
# coding: utf8
import importlib
SqlBuilder = importlib.import_module("sql-builder").SqlBuilder
DbTables = importlib.import_module("monaledge-db-tables")
Users = DbTables.Users
Categories = DbTables.Categories
Articles = DbTables.Articles
Comments = DbTables.Comments

print(SqlBuilder().create_table(Users(0, 'xxxxxx', '2000-01-01T00:00:00Z', '2000-01-01T00:00:00Z', 'name1', 'url1')))

実行すると以下SQL文を出力する。

create table if not exists Users (id integer not null primary key,address text not null,created text not null,updated text not null,name text not null,icon_image_path text not null);

monaledge-db-tables.py

モナレッジの記事用テーブルを想定してみた。

#!/usr/bin/env python3
# coding: utf8
from dataclasses import dataclass, field
from decimal import Decimal
from datetime import datetime 
#@dataclass(slots=True)
@dataclass
class Users:
    id: int
    address: str = field(metadata={'UK':True})
    created: datetime 
    updated: datetime 
    name: str
    icon_image_path: str
#@dataclass(slots=True)
@dataclass
class Categories:
    id: int
    name: str = field(metadata={'UK':True})
#@dataclass(slots=True)
@dataclass
class Articles:
    id: int
    created: datetime 
    updated: datetime 
    title: str
    sent_mona: Decimal
    access: int
    ogp_path: str
    category_id: int
    content: str
#@dataclass(slots=True)
@dataclass
class Comments:
    id: int
    article_id: int
    created: datetime 
    updated: datetime 
    user_id: int
    content: str

sql-builder.py

DataClassからSQL文を作る。

#!/usr/bin/env python3
# coding: utf8
import dataclasses 
from decimal import Decimal
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
import typing
class SqlBuilder:
    def table_name(self, data):
        if isinstance(data, type): return data.__name__
        else: return type(data).__name__
    def column_names(self, data): return list(data.__dataclass_fields__.keys())
    def column_index(self, data, name): return self.column_names(data).index(name)
    def to_type(self, type): # INTEGER,REAL,TEXT,BLOB,NULL
        if type is int or type is bool: return 'integer'
        elif type is float or type is complex: return 'real'
        elif type is list: return 'blob'
        else: return 'text'
    def to_const(self, f): # primary key, unique, not null, check(), default
        consts = ['not null']
        if 'id' == f.name: consts.append('primary key')
        if f.type is bool: consts.append(f"check({f.name} = 0 or {f.name} = 1)")
        if f.default is not dataclasses._MISSING_TYPE and type(f.default) is not dataclasses._MISSING_TYPE: consts.append(f"default {f.default}")
        if isinstance(f.metadata, list): consts += [self.expand_metadata(k,v) for k,v in f.metadata.items()]
        return ' '.join(consts)
    def from_iso(self, iso): return datetime.fromisoformat(iso.replace('Z', '+00:00'))
    def quote(self, v):
        t = type(v)
        if t is int or t is float: return str(v)
        elif t is bool: return str(1 if v else 0)
        elif t is datetime: return f"'{v.isoformat().replace('+00:00', 'Z')}'"
        else: return f"'{v}'"
    def expand_metadata(self, k, v):
        match key:
            case 'PK': return 'primary key'
            case 'UK': return 'unique'
            case 'FK': 
                w = f.metadata[key].split(' ')
                return f'references {w[1]}({w[2]})'
            case 'CK': return f'check ({f.metadata[key]})'
            case 'NOW': return f'default current_timestamp'
            case _: return v
    def create_table(self, data):
        columns = []
        for field in data.__dataclass_fields__.values():
            columns.append(' '.join([field.name, self.to_type(field.type), self.to_const(field)]))
        return f"create table if not exists {self.table_name(data)} ({','.join(columns)});"
    def insert(self, data):
        return f"insert into {self.table_name(data)} values ({','.join([getattr(data, key) for k in data.__dataclass_fields__.keys() if type(v) is not dataclasses._MISSING_TYPE])});"
    def update(self, data, where):
        values = ','.join([f'{k} = {self.quote(v)}' for k,v in data.__dataclass_fields__.items() if type(v) is not dataclasses._MISSING_TYPE])
        conds = ','.join([f'{k} = {self.quote(v)}' for k,v in where.__dataclass_fields__.items() if type(v) is not dataclasses._MISSING_TYPE])
        return f"update {self.table_name(data)} set {values} where {conds};"
    def delete(self, where):
        conds = ','.join([f'{k} = {self.quote(v)}' for k,v in where.__dataclass_fields__.items() if type(v) is not dataclasses._MISSING_TYPE])
        return f"delete from {self.table_name(data)} where {conds};"
    def clear(self): return f"delete from {self.table_name(data)};"
    def _get_exists_kvs(self, data): return [(k,v) for k,v in data.__dataclass_fields__.items() if type(v) is not dataclasses._MISSING_TYPE]
    def _get_exists_keys(self, data): return [k for k,v in data.__dataclass_fields__.items() if type(v) is not dataclasses._MISSING_TYPE]
    def _get_exists_values(self, data): return [v for k,v in data.__dataclass_fields__.items() if type(v) is not dataclasses._MISSING_TYPE]
    def insert_preper(self, data): # sqlite3.exec(sql, params)の引数をタプルで返す
        ks = ','.join(list('?' * len(data.__dataclass_fields__.keys())))
        vs = self._get_exists_values(data)
        return (f"insert into {self.table_name(data)}({','.join(ks)}) values ({','.join(vs)})", vs)
    def update_preper(self, data, where): # sqlite3.exec(sql, params)の引数をタプルで返す
        dks = self._get_exists_keys(data)
        dvs = self._get_exists_values(data)
        wks = self._get_exists_keys(where)
        wvs = self._get_exists_values(where)
        return (f"update {self.table_name(data)} set {','.join([f'{k} = ?' for k in dks])} where {','.join([f'{k} = ?' for k in wks])};", dvs + wvs)

問題

DataClassの型変換
- 日付型
  - Pythonのdatetime.fromisoformat()では末尾Zの日時が変換できない
  - SQLite3の日付は末尾Zが基本
  - ISO8601の仕様でも末尾Zが基本

+00:00ならZと同じ意味になるしdatetime.fromisoformat()でも変換できる。でも-00:00とも書けるし、+00:00:00.000とも書ける。つまり表記ゆれがある。

外部からISO文字列で日付を取得するとき、以下のように変換しないといけない。

iso = '2000-01-01T00:00:00Z'
pyiso = iso.replace('Z', '+00:00')

ISO仕様を満たしていないのにISOを名乗るのはどうかと思うよPythonさん。

PythonのDataClassでORMっぽいのを書いてみる

ブツ