PythonのSQLクエリビルダーPyPikaで動的カスタムSQL文生成方法を考える
TL:DR
-
pypika.terms.Termを継承したクラスを作成 - 複数単語のリスト(
Type | Any)をインスタンス引数に渡す -
__call__()で変数値を辞書として引数に渡し、クラスの型(Type)の単語をそれらで置換したインスタンスを返却 -
get_sql()で変数値のうちget_sql()が呼べるならget_sql()、それ以外は__str__()を呼び出して" ".join()で結合
背景
PyPikaは一般的なINSERTやUPDATEといった基本的なCRUD等はサポートしていて、以下のように記述できる。
from pypika import PostgreSQLQuery as Query, Table
table: Table = Table("sample_table")
print(Query.update(table).set(table.age, 20).where(table.id == 40).get_sql())
---
UPDATE "sample_table" SET "age"=20 WHERE "id"=40
しかし、言語共通的な文はこのように記述できる一方でRDBMS固有のSQL文にはほとんど対応していない。
一応関数であればCustomFunctionというクラスを用いることで、自前で関数内容を定義して動的に関数の引数を渡すことは可能である。
from pypika import CustomFunction
customers = Tables('customers')
DateDiff = CustomFunction('DATE_DIFF', ['interval', 'start_date', 'end_date'])q = Query.from_(customers).select(
customers.id,
customers.fname,
customers.lname,
DateDiff('day', customers.created_date, customers.updated_date)
)
だが例えば以下のようなPostgreSQLのインデックスを作成するSQL文は上記のような書き方ができない。(最後に必ず()が付いてしまうため)
CREATE INDEX "name" ON "table"
確認した限りでは関数の引数でないインデックス名やテーブル名といった単語の部分を動的に変更したい場合、それに対応した仕組みは提供されていないようである。
なおCREATE文をサポートしているのはテーブル作成しかない。
そこで、関数形式以外であっても引数を渡してカスタムSQL文を動的に生成するための仕組みを考えてみる。
方針
単純に例えばf'CREATE INDEX "{name}" ON "{table}"'を返すだけといったような関数を定義しても良いが、カスタムSQL文を作るたびにdefやlambdaが増えるのはあまりスマートではないし、定義のたびにミスが混入する可能性が比較的多そうなどメンテナンス性も良くない。
また、Queryクラスでクエリ構築した際の戻り値QueryBuilderクラスは、SQL文を出力するget_sql()に引数を渡すことで出力をカスタマイズすることができるが、文字列を返却するだけの関数を用意してしまうとこれと一緒のコレクションに集約して一括でget_sql()を呼び出すことができないので、使い勝手が良いとは言えない。
そこで、QueryBuilderの継承元でありget_sql()を定義しているTermクラスを継承し、引数としてSQL文の単語群にはクラスの型やそのインスタンスも渡せるようなクラスを作成する。
クラスの型の単語は後から変数のように見なして動的な置換対象とする。
置換方法はインスタンス()のように呼ぶことで動的部分を呼び出し時の引数で置換し、置換後の継承クラスを返す。
そして継承クラスのget_sql()呼び出し時には、単語群それぞれがget_sql()を呼べるならget_sql()、それ以外は__str__()を呼び出して" "で結合する。
これにより、既存のQueryBuilderと部分的に互換を保ちつつ、かつPyPikaの仕組みを可能な限り生かすことができる。
※単語群の型はType | Anyとしている。Term継承クラスのみを単語群に指定しても良かったが、get_sql()を継承無しで直接実装しているTableクラスがあったり定義もしていないSchemaクラスもあったりするなどPyPikaの詳細設計を網羅しきれないので、SQL文作成時にランタイムでget_sql()が呼べるかどうかで判定している。
※新規クラスはQueryBuilderを継承しても良かったが、逆にQueryBuilderの仕組みを用いて誤った使い方をすることで、自前定義の想定範囲を超えたあり得ないSQL文をいくらでも構築可能になってしまうので不採用とした。
実装
以下のようなクラスを作成する。
import copy
from typing import Any, Dict, List, Optional, Self, Sequence, Type, Union
from pypika.terms import Term
from pypika.utils import QueryException
class MultiTerm(Term):
def __init__(
self,
terms: Sequence[Union[Type, Any]],
var_idxs: Optional[Sequence[int]] = None,
opt_idxs: Optional[Sequence[int]] = None,
opt_enables: Optional[Sequence[int]] = None,
get_sql_kwargs: Optional[Dict[int, Dict[str, Any]]] = None,
) -> None:
self.terms: List[Union[Type, Any]] = list(terms)
self.var_idxs: List[int] = list(var_idxs or [])
self.var_idxs.extend(
[
i
for i, t in enumerate(self.terms)
if isinstance(t, type) and i not in self.var_idxs
]
)
self.opt_idxs: List[int] = [
i for i in (opt_idxs or []) if i in range(len(self.terms))
]
self.opt_enables: List[int] = [
e for e in (opt_enables or []) if e in self.opt_idxs
]
self.get_sql_kwargs: Dict[int, Dict[str, Any]] = copy.deepcopy(
get_sql_kwargs or {}
)
def __call__(self, *args: Any, **kwds: Any) -> Self:
new_terms: List[Union[Type, Any]] = list(self.terms)
if 0 < len(self.var_idxs) and 0 < len(args) and isinstance(args[0], dict):
for i, sql_var in args[0].items():
if i not in range(len(new_terms)) or i not in self.var_idxs:
continue
current_term: Union[Union[Type, Any]] = new_terms[i]
if not isinstance(sql_var, (str, int, float)):
if isinstance(current_term, type) and not issubclass(
sql_var.__class__, current_term
):
raise QueryException(
f"variable is not an expected subclass. "
f"actual: {sql_var.__class__}({sql_var}) "
f"expected: {current_term}"
)
elif not isinstance(current_term, type) and not isinstance(
sql_var, current_term.__class__
):
raise QueryException(
f"variable is not an expected instance. "
f"actual: {sql_var.__class__}({sql_var}) "
f"expected: {current_term.__class__}({current_term})"
)
new_terms[i] = sql_var
return MultiTerm(
new_terms,
var_idxs=self.var_idxs,
opt_idxs=self.opt_idxs,
opt_enables=self._make_opt_enables(**kwds),
get_sql_kwargs={
**(copy.deepcopy(self.get_sql_kwargs)),
**(kwds.get("kwargs", {})),
},
)
def _make_opt_enables(
self,
opt_enables: Optional[Sequence[int]] = None,
opt_disables: Optional[Sequence[int]] = None,
**_,
) -> List[int]:
new_enables: List[int] = list(self.opt_enables)
new_enables.extend([e for e in (opt_enables or []) if e not in new_enables])
return [e for e in new_enables if e not in (opt_disables or [])]
def get_sql(self, **kwargs: Any) -> str:
kw_args: Dict[int, Dict[str, Any]] = kwargs.get("kwargs", {})
opt_enables: List[int] = self._make_opt_enables(**kwargs)
sql_strs: Sequence[str] = []
for i, term in enumerate(self.terms):
if i in self.opt_idxs and i not in opt_enables:
continue
elif i not in self.var_idxs:
sql_strs.append(
term.get_sql() if self._callable_get_sql(term) else str(term)
)
elif self._callable_get_sql(term):
sql_strs.append(
term.get_sql(
**(
{
**((self.get_sql_kwargs or {}).get(i, {})),
**kwargs,
**(kw_args.get(i, {})),
}
)
)
)
elif not isinstance(term, type):
sql_strs.append(str(term))
else:
raise QueryException(
f"term is still a type. "
f"replace with an instance of this, or str | int: {term}"
)
return " ".join(sql_strs)
def _callable_get_sql(self, value: Any) -> bool:
return hasattr(value, "get_sql") and callable(value.get_sql)
基本使用方法
変数化したい位置にNode継承クラスの型を記述してリストで渡し、ベースのインスタンスを生成する。
from pypika import Field, Table
create_index: MultiTerm = MultiTerm(["CREATE", "INDEX", Field, "ON", Table])
変数の位置に値をあてはめたいときはインスタンス({インデックス: インデックス位置のクラスインスタンス})のように呼び出す。
あるいは({インデックス: str | int})でも良い。この場合str()された文字列がそのまま出力される。
print(
create_index(
{2: Field("field_1"), 4: Table("table_1", schema="schema_1")}
).get_sql()
)
print(create_index({2: "field_2", 4: "table_2"}).get_sql())
---
CREATE INDEX field_1 ON "schema_1"."table_1"
CREATE INDEX field_2 ON table_2
get_sql()のオプション引数をインスタンスに設定
例では2番目の位置の変数に対してget_sql()にオプション引数quote_char='"'を渡すことでField.quote_sql(quote_char='"')と同等の出力結果が得られ、単語が"で囲われる。
create_index: MultiTerm = MultiTerm(
["CREATE", "INDEX", Field, "ON", Table], get_sql_kwargs={2: dict(quote_char='"')}
)
print(create_index({2: Field("field_3"), 4: Table("table_3")}).get_sql())
---
CREATE INDEX "field_3" ON "table_3"
get_sql()呼び出し時にオプション引数を設定
クラスインスタンスにはオプション引数を渡さずに関数呼び出しの時に渡す。
create_index: MultiTerm = MultiTerm(["CREATE", "INDEX", Field, "ON", Table])
print(
create_index({2: Field("field_4"), 4: Table("table_4")}).get_sql(
kwargs={2: dict(quote_char='"')}
)
)
---
CREATE INDEX "field_4" ON "table_4"
なお、インスタンス時に指定したオプションよりも優先される。
下記例では2番目をインスタンス宣言時に"で囲うようにしているがget_sql()呼び出し時のkwargs={2: dict(quote_char="'")}が優先されて'で囲うようになっている。
create_index: MultiTerm = MultiTerm(
["CREATE", "INDEX", Field, "ON", Table], get_sql_kwargs={2: dict(quote_char='"')}
)
print(
create_index({2: Field("field_5"), 4: Table("table_5")}).get_sql(
kwargs={2: dict(quote_char="'")}
)
)
---
CREATE INDEX 'field_5' ON "table_5"
全変数に対して同じオプション引数を渡すことも可能。
⇔変数として判定されていない不変部分(CREATE等)は適用されない。
例ではquote_char="'"が2,4の変数に対してのみ適用される(4には効果なし)。
create_index: MultiTerm = MultiTerm(
["CREATE", "INDEX", Field, "ON", Table], get_sql_kwargs={2: dict(quote_char='"')}
)
print(create_index({2: Field("field_6"), 4: Table("table_6")}).get_sql(quote_char="'"))
---
CREATE INDEX 'field_6' ON "table_6"
また、全変数に対するオプション引数よりも個別のオプション引数のほうが優先される。
create_index: MultiTerm = MultiTerm(["CREATE", "INDEX", Field, "ON", Table])
print(
create_index({2: Field("field_7"), 4: Table("table_7")}).get_sql(
quote_char='"', kwargs={2: dict(quote_char="'")}
)
)
---
CREATE INDEX 'field_7' ON "table_7"
変数置換時にget_sql()のオプション引数を設定
kwargsという名前のオプション引数に指定する。
create_index: MultiTerm = MultiTerm(["CREATE", "INDEX", Field, "ON", Table])
print(
create_index(
{2: Field("field_8"), 4: Table("table_8")}, kwargs={2: dict(quote_char='"')}
).get_sql()
)
---
CREATE INDEX "field_8" ON "table_8"
デフォルト値の指定
インスタンス引数にTerm継承クラスの型ではなくインスタンスを渡す。
変数部分は既定ではTerm継承クラスの型かどうかで判断しているので、追加でどのインデックスが変数かを伝えるためvar_idxsでその位置のリストを渡す。
create_index: MultiTerm = MultiTerm(
["CREATE", "INDEX", Field("default"), "ON", Table("default")], var_idxs=[2, 4]
)
全てデフォルト値が設定されていればインスタンス変数をそのままget_sql()することもでき、またインスタンス().get_sql()でも同じ結果となる。どちらもデフォルト値が出力される。
上書きしたいときはインスタンス()の引数に渡す。
print(create_index.get_sql())
print(create_index().get_sql())
print(create_index({2: Field("field_9")}).get_sql())
---
CREATE INDEX default ON "default"
CREATE INDEX default ON "default"
CREATE INDEX field_9 ON "default"
オプション単語の指定
インスタンスの引数にopt_idxsでどのインデックスがオプションかを指定する。
例ではCONCURRENTLYをオプション単語としている。
create_index: MultiTerm = MultiTerm(
["CREATE", "INDEX", "CONCURRENTLY", Field, "ON", Table], opt_idxs=[2]
)
何も指定しないと以下。
print(create_index({3: Field("field_10"), 5: Table("table_10")}).get_sql())
---
CREATE INDEX field_10 ON "table_10"
オプションを有効にするには、インスタンス()の引数にopt_enablesで指定する。
print(
create_index(
{3: Field("field_11"), 5: Table("table_11")}, opt_enables=[2]
).get_sql()
)
---
CREATE INDEX CONCURRENTLY field_11 ON "table_11"
get_sql()の引数でも指定可能。
print(
create_index({3: Field("field_12"), 5: Table("table_12")}).get_sql(opt_enables=[2])
)
---
CREATE INDEX CONCURRENTLY field_12 ON "table_12"
クラスの型がオプション対象の場合は未指定でもエラーにならない。
create_index: MultiTerm = MultiTerm(
["CREATE", "INDEX", Field, "ON", Table], opt_idxs=[2, 4]
)
print(create_index.get_sql())
---
CREATE INDEX ON
NG例
変数位置の単語が置換されていない
create_index: MultiTerm = MultiTerm(["CREATE", "INDEX", Field, "ON", Table])
print(create_index.get_sql())
pypika.utils.QueryException: term is still a type. replace with an instance of this, or str | int: <class 'pypika.terms.Field'>
同一クラスでもサブクラスでもないクラスで置換
create_index: MultiTerm = MultiTerm(["CREATE", "INDEX", Field, "ON", Table])
print(create_index({2: Field("field_13"), 4: Field("ng_table")}).get_sql())
pypika.utils.QueryException: variable is not an expected subclass. actual: <class 'pypika.terms.Field'>("ng_table") expected: <class 'pypika.queries.Table'>
str | int | floatなら可能。
デフォルト値と異なるインスタンスクラスで置換
create_index: MultiTerm = MultiTerm(
["CREATE", "INDEX", Field("default"), "ON", Table("default")], var_idxs=[2, 4]
)
print(create_index({4: Field("ng_table")}).get_sql())
pypika.utils.QueryException: variable is not an expected instance. actual: <class 'pypika.terms.Field'>("ng_table") expected: <class 'pypika.queries.Table'>("default")
str | int | floatなら可能。
Discussion