やってみる

アウトプットすべく己を導くためのブログ。その試行錯誤すらたれ流す。

SQLite3のトランザクション用フレームワークを作った

ついにそれっぽいのができた。

成果物

GitHubPython.Sqlite3.class.201803041400000

概要

  • トランザクション処理begin(), commit()と、DB接続切断処理dataset.connect(), db.engin.dispose()を省略できる。
  • 複数DBにおいて、任意の操作を、任意のクラスで実装できる

ソースコード

呼出側。ビジネスロジックを呼び出す。Initialize, Add, Remove, Gets

run.py

from db.Service import Service
if __name__ == '__main__':
    # CREATE TABLE
    Service().Main.Initialize()
    # INSERT, DELETE
    id1 = Service().Main.Add('http://1', 'memo1')
    id2 = Service().Main.Add('http://2', 'memo2')
    id3 = Service().Main.Add('http://3', 'memo3')
    Service().Main.Remove(id2)
    # SELECT
    for record in Service().Main.Gets():
        print(record)

ビジネスロジック実装側。./db/service/{任意サービス名}MainService.pyファイルを書けばService().{任意サービス名}属性値としてセットされる。

MainService.py

class MainService:
    def __init__(self):
        self.__dataset_kwargs = {'Urls':{'url': 'sqlite:///Urls.db'}, 'Memos':{'url': 'sqlite:///Memos.db'}}
    @property
    def DatasetArgs(self): return self.__dataset_kwargs
    def Initialize(self, dbs):
        for name in ['Urls', 'Memos']:
            if name in dbs[name].tables:
                dbs[name].query(f'DROP TABLE {name};')
        dbs['Urls'].query('CREATE TABLE Urls (Id integer PRIMARY KEY, Name text);')
        dbs['Memos'].query('CREATE TABLE Memos (Id integer PRIMARY KEY, UrlId NOT NULL, Memo text);')
    def Gets(self, dbs):
        dbs['Urls'].query('attach "{}" as _U;'.format(self.__GetDbPath(self.__dataset_kwargs['Urls']['url'])))
        dbs['Urls'].query('attach "{}" as _M;'.format(self.__GetDbPath(self.__dataset_kwargs['Memos']['url'])) )
        return dbs['Urls'].query('SELECT u.Id, u.Url, m.Memo FROM main.Urls u INNER JOIN _M.Memos m ON u.Id = m.UrlId;')
    def Add(self, dbs, url, memo):
        _id = dbs['Urls']['Urls'].insert({'Url': url})
        dbs['Memos']['Memos'].insert({'UrlId': _id, 'Memo': memo})
        return _id
    def Remove(self, dbs, id):
        dbs['Memos']['Memos'].delete(UrlId=id)
        dbs['Urls']['Urls'].delete(Id=id)

    def __GetDbPath(self, db_url): return db_url.replace('sqlite:///', '')

複数DBを操作できる。DB操作メソッドの引数dbsにはdataset.connect()のdictが渡される。

SQLiteでは複数DBの結合にATTACH文を使う。Getsメソッドでは、UrlsMemos、の2つのDBと、同名のテーブルを結合している。

以下が本質。トランザクションとDB接続切断のフレームワーク部分。

Transactioner.py

class Transactioner:
    def Transact(self, *args, **kwargs):
        dbs = {}
        for key in self.__dataset_kwargs:
            dbs[key] = dataset.connect(**self.__dataset_kwargs[key])
            dbs[key].begin()
        res = self.__transaction(dbs, *args, **kwargs)
        for key in self.__dataset_kwargs:
            dbs[key].commit()
        for key in self.__dataset_kwargs:
            dbs[key].engine.dispose()
            del dbs[key]
        return res

DBが複数あるので複雑化した。with文が使えないから。

また、datasetはDB接続メソッドがない。SQLArchemyのインタフェースにはあるので、さかのぼって参照した。(dbs[key].engine.dispose())

開発環境

前回まで

前回は単一DBごとにしか操作できなかった。今回で改善された。

課題

  • DB接続用引数の指定が面倒
    • self.__dataset_kwargs = {'Urls':{'url': 'sqlite:///Urls.db'}, 'Memos':{'url': 'sqlite:///Memos.db'}}
  • 名付けが面倒(一意チェック必要)
    • DBファイルパス
    • DBキー名
    • ATTACH時のDB名

できるだけ多くの状況に対応するためには、現状が最善だと思うが。DBのパスや名前の管理層はあったほうがいいかもしれない。

所感

いい感じ。こういう風にやりたかったんだよ。