静的型付け#
psycopg のソースコードは、PEP 0484 の型ヒントに従ってアノテーションされていて、Mypy の現在のバージョンで --strict モードを使用して型チェックされています。
アプリケーションが Mypy を使用して型チェックされている場合、psycopg の型を活用して psycopg オブジェクトとデータベースから返されたデータが正しく使用されているかを検証できます。
ジェネリック型#
psycopg の Connection と Cursor オブジェクトは Generic オブジェクトであり、返されたレコードの型である Row パラメータをサポートしています。
デフォルトでは、Cursor.fetchall() などのメソッドは、サイズとコンテンツが未知の通常のタプルを返します。したがって、connect() 関数は型 psycopg.Connection[Tuple[Any, ...]] のオブジェクトを返し、Connection.cursor() は型 psycopg.Cursor[Tuple[Any, ...]] のオブジェクトを返します。ジェネリックな配管コード (訳注: データベースに接続するコード) を書いている場合は、Connection[Any] や Cursor[Any] などのアノテーションを使うのが実用的かもしれません。
conn = psycopg.connect() # 型は psycopg.Connection[Tuple[Any, ...]] です
cur = conn.cursor() # 型は psycopg.Cursor[Tuple[Any, ...]] です
rec = cur.fetchone() # 型は Optional[Tuple[Any, ...]] です
recs = cur.fetchall() # 型は List[Tuple[Any, ...]] です
返された行の型#
データを異なる型、たとえばディクショナリとして返すコネクションとカーソルを使用したい場合、connect() と cursor() メソッドの row_factory 引数が使えます。この引数は、カーソルの fetch メソッドから返されるレコードの型をコントロールして、返されたオブジェクトを適切にアノテートします。詳細については 行ファクトリ を参照してください。
dconn = psycopg.connect(row_factory=dict_row)
# dconn の型は psycopg.Connection[Dict[str, Any]] です
dcur = conn.cursor(row_factory=dict_row)
dcur = dconn.cursor()
# dcur の型はいずれの場合も psycopg.Cursor[Dict[str, Any]] です
drec = dcur.fetchone()
# drec の型は Optional[Dict[str, Any]] です
例: レコードを Pydantic モデルとして返す#
Pydantic を使用すると、ランタイム時の静的型付けの矯正が可能になります。Pydantic モデル ファクトリ を使用すると、Mypy を使用してコードの静的な型チェックができ、返された行がモデルと互換性がない場合にデータベースのクエリが例外を起こすようになります。
次の例では、mypy --strict を使用して問題のレポートなしで型チェックができています。Pydantic は、互換性のないデータを返すクエリで Person が使用された場合に、ランタイム時のエラーも起こします。
from datetime import date
from typing import Optional
import psycopg
from psycopg.rows import class_row
from pydantic import BaseModel
class Person(BaseModel):
id: int
first_name: str
last_name: str
dob: Optional[date]
def fetch_person(id: int) -> Person:
with psycopg.connect() as conn:
with conn.cursor(row_factory=class_row(Person)) as cur:
cur.execute(
"""
SELECT id, first_name, last_name, dob
FROM (VALUES
(1, 'John', 'Doe', '2000-01-01'::date),
(2, 'Jane', 'White', NULL)
) AS data (id, first_name, last_name, dob)
WHERE id = %(id)s;
""",
{"id": id},
)
obj = cur.fetchone()
# ここで reveal_type(obj) は 'Optional[Person]' を返すはずです
if not obj:
raise KeyError(f"person {id} not found")
# ここで reveal_type(obj) は 'Person' を返すはずです
return obj
for id in [1, 2]:
p = fetch_person(id)
if p.dob:
print(f"{p.first_name} さんは {p.dob.year} に生まれました")
else:
print(f"誰も {p.first_name} さんがいつ生まれたのかを知りません")
クエリ内のリテラル文字列の型チェック#
execute() や類似のメソッドのようなメソッドは、PEP 675 によれば、リテラル文字列のみを入力として受け取る必要があります。つまり、クエリは任意の文字列式ではなく、コード中のリテラル文字列から来る必要があるということです。
たとえば、クエリへの引数の引き渡しは execute() の2つ目の引数を経由で行なわれる必要があり、文字列の構築によって行われてはいけません。
def get_record(conn: psycopg.Connection[Any], id: int) -> Any:
cur = conn.execute("SELECT * FROM my_table WHERE id = %s" % id) # BAD!
return cur.fetchone()
# この関数は、以下のように実装する必要があります
def get_record(conn: psycopg.Connection[Any], id: int) -> Any:
cur = conn.execute("select * FROM my_table WHERE id = %s", (id,))
return cur.fetchone()
クエリを動的に構築する場合、sql.SQL と同様のオブジェクトを使用して、テーブルとフィールドの名前を安全にエスケープする必要があります。SQL() オブジェクトのパラメータはリテラル文字列である必要があります。
def count_records(conn: psycopg.Connection[Any], table: str) -> int:
query = "SELECT count(*) FROM %s" % table # BAD!
return conn.execute(query).fetchone()[0]
# この関数は、以下のように実装する必要があります
def count_records(conn: psycopg.Connection[Any], table: str) -> int:
query = sql.SQL("SELECT count(*) FROM {}").format(sql.Identifier(table))
return conn.execute(query).fetchone()[0]
これを書いている時点では、この型チェックを実装している Python の静的型チェッカーは存在しません (mypy はこれを実装していません。Pyre は実装していますが、psycopg ではまだ機能しません)。型チェッカーのサポートが完了したら、上記の悪いステートメントはエラーとして報告されるようになるはずです。