sqlalchemyのautomapによる既存のDBからのmodelの生成について
昨日あたりにsqlalchemyのautomapの仕組みをつかってgraphqlのschemaを作ろうとしていたのだけれど。その時に使っていたsqlalchemyのautomapの仕組みが期待していたのとちょっと違っていたのでどうしようか考えてみる。
sqlalchemy?
pythonで使われているORM mapper的なもの。これ
automap?
古くはsqlsoupと呼ばれる、既存のDBの情報を取り出して、ORM用のmodelを良い感じに作ってくれるメタプログラミングを利用したパッケージがあった。このパッケージにあった機能の概念ををsqlalchemyの作者がsqlalchemyに取り込んだもの。使い方はドキュメントを読むと良い。
automapの仕組み
automapの仕組みは大雑把に言うと以下のもの。
- metadataのreflectを使う
- reflectによって得られた設定を利用してmodelを自動で定義
metadataのreflect
sqlalchemyは接続するdatabaseの情報をmetadataというオブジェクトに格納して管理している(詳しくはこのあたり。
そしてこのmetadataのreflectionの機能を使うと、接続したdatabaseの情報を良い感じで取り出してくれる。以下の様な感じ。
例えば、以下のようなテーブルを持つsqlite3のdatabaseから情報を取り出してみる。
person.db
CREATE TABLE person ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, father_id INTEGER, mother_id INTEGER, FOREIGN KEY(father_id) REFERENCES person(id), FOREIGN KEY(mother_id) REFERENCES person(id) );
sqlite3のdbの情報を取り出した結果。
from sqlalchemy import create_engine, MetaData from pprint import pprint engine = create_engine("sqlite:///person.db") metadata = MetaData() metadata.reflect(engine) pprint(vars(metadata))
こんな感じでdatabaseの情報が取り出せる。
{'_bind': None,
'_fk_memos': defaultdict(<class 'list'>,
{('person', 'id'): [ForeignKey('person.id'),
ForeignKey('person.id')]}),
'_schemas': set(),
'_sequences': {},
'naming_convention': immutabledict({'ix': 'ix_%(column_0_label)s'}),
'schema': None,
'tables': {
'person': Table('person', MetaData(bind=None),
Column('id', INTEGER(), table=<person>, primary_key=True, nullable=False),
Column('name', TEXT(), table=<person>, nullable=False),
Column('father_id', INTEGER(), ForeignKey('person.id'), table=<person>),
Column('mother_id', INTEGER(), ForeignKey('person.id'), table=<person>), schema=None),
'sqlite_sequence': Table('sqlite_sequence', MetaData(bind=None), Column('name', NullType(), table=<sqlite_sequence>),
Column('seq', NullType(), table=<sqlite_sequence>), schema=None)}}
automapはこういう感じで取り出したテーブルの情報を使ってmodelを作る。
automapでmetadataからmodelを作る
これは実質automapのprepareが担っている。この時relationshipの情報も補ってくれるので嬉しい。relationshipの検知の手順はドキュメントに書かれている。すごく雑に言うとForeignKeyのconstraintの情報を使ってrelationshipを検知している。ひどく真っ当で当たり前っぽい。
実際先程のdatabaseの情報からmodel用のクラスを作ってみると以下の様な感じになる。
from sqlalchemy import create_engine from sqlalchemy.ext.automap import automap_base from sqlalchemy.inspection import inspect Base = automap_base() engine = create_engine("sqlite:///person.db") Base.prepare(engine, reflect=True) # 生成されたmodel print(Base.classes.person) # modelが持つfield(property)を利用するためにinspectionの機能を使っている mapper = inspect(Base.classes.person) for prop in mapper.iterate_properties: print("\t", prop.key, type(prop))
結果はこういう感じ。person,person_collectionが検知されたrelationship。 それ以外はtableの定義で存在していたfield。
<class 'sqlalchemy.ext.automap.person'>
person <class 'sqlalchemy.orm.relationships.RelationshipProperty'>
person_collection <class 'sqlalchemy.orm.relationships.RelationshipProperty'>
id <class 'sqlalchemy.orm.properties.ColumnProperty'>
name <class 'sqlalchemy.orm.properties.ColumnProperty'>
father_id <class 'sqlalchemy.orm.properties.ColumnProperty'>
mother_id <class 'sqlalchemy.orm.properties.ColumnProperty'>
automapの問題
真っ当だと思っていたautomapによるrelationshipの導出にも限界がある。よーく先程の結果を見るとそれが分かるかもしれない。
personとperson_collectionが導出されているけれど。これはどのforeignKeyと結びついているのだろう？
実際前回の記事で作ったコード(gen.pyの方)を利用して、dbから取り出してみた情報をみると以下のような結果になる。
{ "person": { "father_id": { "type": "Integer", "nullable": true }, "id": { "type": "ID", "nullable": false }, "mother_id": { "type": "Integer", "nullable": true }, "name": { "type": "String", "nullable": false }, "person": { "table": "person", "direction": "MANYTOONE", "uselist": false, "relation": { "from": "person.id", "to": "person.father_id" } }, "person_collection": { "table": "person", "direction": "ONETOMANY", "uselist": true, "relation": { "from": "person.id", "to": "person.mother_id" } } } }
relationのfromとtoのところを見て欲しい。toがfather_idだったりmother_idだったりしている。そしてこれは実行する度にコロコロ変わる。 もう少し正確に言うなら、automapのrelationshipの導出の機能は、同じテーブルに対する参照のfoerignkey constraintを持っていた場合に、これを上手く取り扱う事ができない。
これをどうにかしないとだめ。
導出されるrelationshipの名前の決め方(もう少し詳細な話)
automapのprepareはそれなりに良く出来ていて、色々挙動を変える事ができる。良い。
上でのrelationshipの導出が上手くいかない現象が起きていた原因をもう少し整理すると以下の様な形になっている。
- foreign keyのconstraintをiterateする
- 各constraintからrelationshipの名前を取り出す(person,person_collectionが返る)
- 取り出した名前を利用してrelationshipを作成する
- この時取り出した名前が重複していた(iterationは辞書に対するものなのでunordered)
というわけで、名前を重複させないように変えてあげれば良い。デフォルトの名前を決める実装は以下のようなもの。
def name_for_scalar_relationship(base, local_cls, referred_cls, constraint): return referred_cls.__name__.lower() def name_for_collection_relationship( base, local_cls, referred_cls, constraint): return referred_cls.__name__.lower() + "_collection"
それぞれcollection(複数)の場合scalar(単数)の場合の名前を生成する処理のデフォルト実装。クラス名(おおよそテーブル名)と対応するものになっているので、当然同じテーブルを参照するconstraintからは同じ名前のrelationshipが生成されてしまう。
衝突しないrelationship名の決め方
衝突しないrelationship名を生成するには、foreignKeyの名前を見るようにすると良いかもしれない(このあたりになるとad-hocな感じになるし。汎用的な方法というのは夢の彼方のようなものなのかもな〜みたいな儚さを感じる)。
例えば、上の例でいうとmother_id,father_idという名前のforeign keyだったので、mother,fatherみたいなrelationshipが作られると良いかもしれない。mother_id - _id = mother みたいな感じ。
def name_for_scalar_relationship(base, local_cls, referred_cls, constraint): basename = constraint.column_keys[0].replace("_id", "") return basename def name_for_collection_relationship(base, local_cls, referred_cls, constraint): basename = constraint.column_keys[0].replace("_id", "") return basename + "_collection"
これをprepareに渡す。
Base.prepare(
engine,
reflect=True,
name_for_scalar_relationship=name_for_scalar_relationship,
name_for_collection_relationship=name_for_collection_relationship,
)
生成される結果は以下のようになる。diffだけ表示。
--- /tmp/before.json +++ /tmp/after.json @@ -1,5 +1,23 @@ { "person": { + "father": { + "direction": "MANYTOONE", + "relation": { + "from": "person.id", + "to": "person.father_id" + }, + "table": "person", + "uselist": false + }, + "father_collection": { + "direction": "ONETOMANY", + "relation": { + "from": "person.id", + "to": "person.father_id" + }, + "table": "person", + "uselist": true + }, "father_id": { "nullable": true, "type": "Integer" @@ -8,15 +26,7 @@ "nullable": false, "type": "ID" }, - "mother_id": { - "nullable": true, - "type": "Integer" - }, - "name": { - "nullable": false, - "type": "String" - }, - "person": { + "mother": { "direction": "MANYTOONE", "relation": { "from": "person.id", @@ -25,14 +35,22 @@ "table": "person", "uselist": false }, - "person_collection": { + "mother_collection": { "direction": "ONETOMANY", "relation": { "from": "person.id", - "to": "person.father_id" + "to": "person.mother_id" }, "table": "person", "uselist": true + }, + "mother_id": { + "nullable": true, + "type": "Integer" + }, + "name": { + "nullable": false, + "type": "String" } } }
personの代わりにmother,fatherというrelationshipが作られた。良い感じ(のようにみえる)。
良い感じに見えたのだけれどまだ足りない(foreign_keyの問題)
さて、先程のforeign keyを使う実装で良い感じにrelationshipが導出できたように見えた。ところでこの方法には問題がある。
例えば以下のような同じ名前のforeign keyを利用して複数のテーブルから参照されているテーブルが有る場合について考えてみる。
あんまりいい例が思いつかないけれど。いろんな言語用のデータがあるようなテーブル(language)があるとする、その行のデータが何言語のデータだったかを調べるためにこのテーブルと結びついているとする。
テキトウにuserテーブルとitemテーブルがあるとする(テキトウでごめんなさい)。
user -> language item -> language
SQLは以下の様な感じ。
CREATE TABLE language ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL ); CREATE TABLE user ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, language_id INTEGER, FOREIGN KEY(language_id) REFERENCES language(id) ); CREATE TABLE item ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, language_id INTEGER, FOREIGN KEY(language_id) REFERENCES language(id) );
このような構造を持ったdbに対して先程の方法でrelationshipを生成しようとしてみたらなんか上手くいかない。
sqlalchemy.exc.ArgumentError: Error creating backref 'language_collection' on relationship 'item.language': property of that name exists on mapper 'Mapper|language|language
エラーになる。
それぞれのrelationshipでの名前と結びつくクラスの名前(テーブル名)を表にしてみると以下の様になる。
| type | name | local_class | referred_class |
|---|---|---|---|
| scalar | language | user | language |
| collection | language_collection | language | user |
| scalar | language | item | language |
| collection | language_collection | language | item |
よく考えたら当たり前で、例えば、languageとuserのjoinを考えると user as u join language as l on u.language_id=l.id みたいな形になる。この時、userからlanguageへの参照はuser.language_id -> languageというように上手くlanguageを取り出せるけれど。その逆はforeign keyの情報からは決定できない。
そしてそもそも先程の手順で使っていたcolumn_keysと言うのは適切な値ではない。常に["language_id"]を返していた(だめじゃん)。
期待する名前は以下の様な感じかも？
| type | name | local_class | referred_class |
|---|---|---|---|
| scalar | language | user | language |
| collection | user_collection | language | user |
| scalar | language | item | language |
| collection | item_collection | language | item |
そんなわけで、単数の方ではcolumn_keysの値を使うのは良いのだけれど。複数の方ではreferred_classを使って欲しい。
正確にいうと、relationship名は以下の２つを勘案して決めて欲しい
- referred_class(referred_table)の名前(真面目にするなら複数形を単数形に直したもの)
- local_class(local_table)の持つforeign keyの名前
例えばこんな感じ。
def guess_name(local_cls, referred_cls, constraint): names = [] arrived = set() itr = itertools.chain( [referred_cls.__name__], [col.name for col in constraint if col.table == local_cls.__table__], ) for name in itr: name = name.replace("_id", "") if name.lower() in arrived: continue arrived.add(name) names.append(name) return "_".join(names) def name_for_scalar_relationship(base, local_cls, referred_cls, constraint): return guess_name(local_cls, referred_cls, constraint) def name_for_collection_relationship(base, local_cls, referred_cls, constraint): return guess_name(local_cls, referred_cls, constraint) + "_collection"
今度は良い感じ。
| type | name | local_class | referred_class |
|---|---|---|---|
| scalar | language | user | language |
| collection | user_collection | language | user |
| scalar | language | item | language |
| collection | item_collection | language | item |
personの方は以下の様な感じになる。
| type | name | local_class | referred_class |
|---|---|---|---|
| scalar | person_father | person | person |
| collection | person_father_collection | person | person |
| scalar | person_mother | person | person |
| collection | person_mother_collection | person | person |
まぁ許容範囲内
many to manyの場合
もうちょっと考えてmany to manyのことも考えてみる。XとYの関連があった時、XtoYという中間テーブル越しに X.id==XtoY.x_id, XtoY.y_id==Y.id というような形。これも同じテーブルと繋がるものが一つだけしかなければtable名を参照で問題ない。一方でXとYがgoodXtoYとbadXtoYで繋がるみたいな場合も考えてみると同様に失敗する。
中間テーブルのことを考えるとこういう感じになるかもしれない。
--- /tmp/guess.before.py 2017-07-01 23:44:10.000000000 +0900 +++ /tmp/guess.after.py 2017-07-01 23:44:06.000000000 +0900 @@ -3,6 +3,7 @@ arrived = set() itr = itertools.chain( [referred_cls.__name__], + [col.table.name for col in constraint if col.table not in (local_cls.__table__, referred_cls.__table__)], [col.name for col in constraint if col.table == local_cls.__table__], ) for name in itr:
こういうテーブルの時に
CREATE TABLE x ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL ); CREATE TABLE y ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL ); CREATE TABLE good_xtoy ( x_id INTEGER, y_id INTEGER, FOREIGN KEY(x_id) REFERENCES x(id), FOREIGN KEY(y_id) REFERENCES y(id) ); CREATE TABLE bad_xtoy ( x_id INTEGER, y_id INTEGER, FOREIGN KEY(x_id) REFERENCES x(id), FOREIGN KEY(y_id) REFERENCES y(id) );
こういう感じになる。
| type | name | local_class | referred_class | intermediate table |
|---|---|---|---|---|
| collection | y_good_xtoy_collection | x | y | good_xtoy |
| collection | y_bad_xtoy_collection | x | y | bad_xtoy |
| collection | x_good_xtoy_collection | y | x | good_xtoy |
| collection | x_bad_xtoy_collection | y | x | bad_xtoy |
名前とかもう少しどうにかしたいなー。
既存のdbからgraphqlのschemaを生成しようとしてみる
はじめに
既存のDBのURLを渡すと、何か良い感じにgraphqlのベースのapiを良い感じに提供してくれるようにする何かを作ろうとしはじめた。 graphqlはschemaを取るのだけれど、こちらのschemaはgraphベースなのでちょっと困る。 サーバー側の実装をするためにはforeignkeyやrelationの情報を知りたいのでいきなりgraphql用のschemaを生成してはだめ。
そんなわけでschemaを作る手前段階の中間的なファイルを生成する。
その後、作った中間表現からgraphql用のschemaを生成してみる。
もくろみ
sqlalchemyのautomapの機能を使うとそれなりに手軽にできるのような気がした。
とりあえず以下の事が全部分かるようなデータを作ると良い。
- 存在するテーブルの情報
- テーブルの持つフィールドの情報
- テーブルの持つ関係(relation)の情報
やってみる
やってみた。あとでまじめに綺麗にするけれど。とりあえずのプロトタイプとしては良い感じ。
以下のようなtableを用意した。
CREATE TABLE childs ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT ); CREATE TABLE kinds ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL ); CREATE TABLE dogs ( id INTEGER PRIMARY KEY AUTOINCREMENT, kind_id INTEGER, dog TEXT, FOREIGN KEY(kind_id) REFERENCES kinds(id) ); CREATE TABLE child_dogs ( child_id INTEGER, dog_id INTEGER, FOREIGN KEY(child_id) REFERENCES childs(id), FOREIGN KEY(dog_id) REFERENCES dogs(id) );
dogsとchildsがchild_dogs経由でmany to many。dogとkindにfkが貼られている。 (database上ではone to oneとone to manyのどちらであるかは決められないので注意)
sqlite上でtableを生成して、生成したdbの情報をgen.pyというファイルに渡す(後述)。
$ cat create.sql | sqlite3 dog.db $ python gen.py 'sqlite:///./dog.db'
以下のような中間表現を得られる。粗削りではあるけれど。
{ "dogs": { "kinds": { "uselist": false, "direction": "MANYTOONE", "type": "kinds", "relation": { "to": "dogs.kind_id", "from": "kinds.id" } }, "id": { "nullable": false, "type": "ID" }, "kind_id": { "nullable": true, "type": "int" }, "dog": { "nullable": true, "type": "str" }, "childs_collection": { "uselist": true, "direction": "MANYTOMANY", "type": "childs", "relation": { "to": "child_dogs.dog_id", "from": "dogs.id" } } }, "kinds": { "id": { "nullable": false, "type": "ID" }, "name": { "nullable": false, "type": "str" }, "dogs_collection": { "uselist": true, "direction": "ONETOMANY", "type": "dogs", "relation": { "to": "dogs.kind_id", "from": "kinds.id" } } }, "childs": { "dogs_collection": { "uselist": true, "direction": "MANYTOMANY", "type": "dogs", "relation": { "to": "child_dogs.child_id", "from": "childs.id" } }, "id": { "nullable": false, "type": "ID" }, "name": { "nullable": true, "type": "str" } } }
存在するテーブルの情報
dogs, childs, kindsのテーブルがあることが分かる
テーブルの持つフィールドの情報
typeがあるものがフィールド。foreign keyとして扱われるものはIDになっている。
テーブルの持つ関係(relation)の情報
relationのfromとtoがわかり、directionも分かるので良さそう。
gen.py
作ったスクリプトは以下の様な感じ。
from collections import OrderedDict from sqlalchemy.ext.automap import automap_base from sqlalchemy import create_engine from sqlalchemy.inspection import inspect def collect(classes, getname=str): d = OrderedDict() for c in classes: mapper = inspect(c) d[mapper.local_table.fullname] = _collect_from_mapper(mapper) return d def _collect_from_mapper(m): d = OrderedDict() for prop in m.iterate_properties: if hasattr(prop, "direction"): pairs = prop.synchronize_pairs assert len(pairs) == 1, "multi keys are not supported" d[prop.key] = { "table": prop.target.fullname, "direction": prop.direction.name, "uselist": prop.uselist, "relation": { "to": "{}.{}".format(pairs[0][0].table.fullname, pairs[0][0].name), "from": "{}.{}".format(pairs[0][1].table.fullname, pairs[0][1].name), } } else: assert len(prop.columns) == 1, "multi keys are not supported" c = prop.columns[0] d[prop.key] = { "type": "ID" if c.primary_key else c.type.python_type.__name__, "nullable": c.nullable, } return d def main(src): Base = automap_base() engine = create_engine(src) Base.prepare(engine, reflect=True) from dictknife import loading d = collect(Base.classes) loading.dumpfile(d, format="json") if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--src", default="sqlite:///./dog.db") args = parser.parse_args() main(args.src)
中間表現からgraphqlのschemaを生成しようとしてみる
先程作った中間表現JSONファイルを利用してgraphqlのschemaを作ってみる。
やってみた結果
名前があんまりよろしくないけれど。以下の様な感じになる。
# gen.json は先程生成した中間表現のJSON $ python convert.py gen.json
このような結果が得られる
type Child {
dogs_collection: [Dog]
id: ID!
name: String
}
type Dog {
kinds: Kind
id: ID!
kind_id: Integer
dog: String
childs_collection: [Child]
}
type Kind {
id: ID!
name: String!
dogs_collection: [Dog]
}
テーブル名から型名を作っているので少し違和感のある名前かも知れない(手抜きをしたかったのでchildrenではなくchildsという名前だった)。 まじめに調べていないので型の書き方が間違っているかもしれないけれど。とりあえずプロトタイプなので。
もう少し後でドキュメントなどを見直す必要がある。このあたり
そう言えば、unionとかenumには対応してない。
コード
コードはこんな感じ。 prestringとdictknifeが必要。
# -*- coding:utf-8 -*- from dictknife import loading from prestring import Module import contextlib import logging logger = logging.getLogger(__name__) def titleize(name): if not name: return name return name[0].upper() + name[1:] def singular(name): if name.endswith("s"): return titleize(name[:-1]) return titleize(name) class Array: def __init__(self, t): self.t = t def __str__(self): return "[{}]".format(self.t) class GraphQLModule(Module): @contextlib.contextmanager def type_(self, name): self.stmt("type {} {{", name) with self.scope(): yield self.stmt("}") def field(self, name, typ, nullable=True): if nullable: self.stmt("{}: {}", name, typ) else: self.stmt("{}: {}!", name, typ) def emit(m, d): for name, fields in d.items(): with m.type_(singular(name)): for k, v in fields.items(): if "type" in v: m.field(k, v["type"], nullable=v.get("nullable", True)) else: if v["uselist"]: m.field(k, Array(singular(v["table"])), nullable=v.get("nullable", True)) else: m.field(k, singular(v["table"]), nullable=v.get("nullable", True)) def main(src): d = loading.loadfile(src) m = GraphQLModule() emit(m, d) print(m) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("src") args = parser.parse_args() main(args.src)
