python + asgi + graphqlまわりの試行錯誤用の最初のコードの作成。あるいは何かを学びたくなったときに最初に何をするかについて

最近はpython + asgi + grpahqlあたりで試行錯誤をすることが少しずつ増えてきた。その辺りのためのコードのメモ(gist)。

ついでに何か新しい領域を学ぼうと思ったとき、あるいは以前触れていたものの陳腐化してしまったとかんじている領域に対する知識のアップデートをしようと思ったときに何をするのかということをメモしようと思った。

graphql?

graphQLはクライアント側で返ってくるレスポンスの形式をクエリーを内に含めてリクエストできるようなエンドポイントを作成できるような何か。詳しくは色々な解説を読み漁ると良い。

  • (暇だったらこの辺りに良さそう資料のURLを貼る)

仕様はこの辺にある。

今回気にするのはpythonでのgraphQLについて。

情報収集 (依存関係の確認)

最初にすることはもちろん情報収集。ただそれぞれの詳しいやり方を調べる前にどのような依存関係があるかを調べることが多い気がする。あとエコシステム。

pythonでのgraphql用のライブラリについて

pythonでのgraphql用のライブラリについては大まかに以下の3つがある。

  • graphene -- 老舗, graphql-coreと同じところが作っている, graphql-coreのまま
  • ariadne -- 新興, asgiのinterface, schema first, graphql-core-nextが使われている
  • strawberry -- こちらも新し目。あまり詳しくない。dataclassesにinspireされているらしい, graphql-core-nextが使われている

もう少しまともな比較はこの辺りを覗くと良い。

依存関係は以下の様な形

graphql-core-next -> ariadne, starwberry
graphql-core -> graphene

graphql-core, graphql-core-next?

pythonでのgraphqlの基礎部分のライブラリには、同じorganizationが作っているが、graphql-core-nextgraphql-coreがあった。

現在はgraphql-core-nextはgraphql-coreに統合され。graphql-core自体はlegacyになった。

$ python -m pip install "graphql-core>=3a"

しかしエコシステム(特にgraphene)が完全にアップデートできているかと言うとそうではない模様。

なぜgraphql-core-nextを作ることになったのかの経緯はこの辺りでも書かれている。python3.6以降の機能を使って書きたいよねとか実装が陳腐化してしまっていて元のGraphQL.jsに追いつけてないとかそのあたりが理由。

graphql-core-nextのリポジトリのreadmeにそれぞれの対応状況についての記述がある。

今回は新しいものを試したいのでgraphql-core-next上のものをみていくことにする。ariadneのあたりを試してみようと思った。

色々な粒度で実行を確認できるようにする

誰しもがある程度の雰囲気を掴んだらhello world的なものを進めていくとは思う。その後の動きは深さ優先的な人と幅優先的な人がいるように思う(雰囲気で話している)。個人的にはそのまま学ぼうとしている何かを使おうとしたり次に直接進む前に、色々な粒度での実行確認ができるような実験場を作ろうすることが多い。

理由は試行錯誤の手間が学習の妨げになるので。何かを確認したいと思った時にはちょうど良い粒度でのコードでの実行確認ができると良い。フットプリントが大きすぎては確認する手間を惜しむ様になっていく。そのような状態では細かな部分への理解が後々おぼつかなくなっていく。あるいは単純に不要な部分を切り分けることができなくなっていくような気がしている。そうなってくると何を学んで何を学ばなかったのかなど記録としての価値が喪われていってしまうような感覚がある。

この辺りについて色々と語りたいこともある気がするけれど。省略。

情報収集で分かったことを雑にまとめる

自分のための備忘録としてはここからが本題。大雑把に分かっていることをまとめる。今回はariadneを対象にした例。

  • pythonのgraphqlライブラリにはgraphene,strawbery,ariadneがある
  • より基礎的なライブラリとしてgraphql-core,graphql-core-nextがある
  • ariadneはgraphql-core-nextの上に構築されている
  • ariadneはasgi用のインターフェイスが用意されている
  • asgiは非同期用のインターフェイスのこと。wsgiの非同期版(wsgiを知らなければwsgiも知っておく)
  • asgiを試すのにはuvicorn辺りが手軽
  • そうは言っても常にasgi込みで実行するのはめんどくさい

1ファイルで実行したい

色々な粒度のものを1ファイルで実行したい。なぜ1ファイルかというとこれから数十回くらいは試すことになるので。そのたびにプロジェクトやらを作っているのはバカバカしい。cookiecutterなどのプロジェクトテンプレートで生成を手軽にしたとしてファイルが不用意に増えていくのは把握がし辛い。

基本は以下のどちらか

$ python <target file>.py
$ pytest <target file> # もしくは単にpytest

引数が欲しくなるときのためにタスクランナーやMakefileなどで指定できるようにしておく。

Makefile

# hello world
00:
  # python 00hello.py とか書くのが面倒なので
    python $@*.py

# aggressive
01:
    python $@*.py --aggressive

# pytest
02:
    pytest $@* -vv

コマンドラインの引数や実行方法を意識せずに色々な例を試したいという気持ちでの作業なのでMakefileである必要はない。他の手に馴染んだものを使えば良い。実際その時の気分で利用するものは変える。

どのような粒度で実行したいか考える

例えば以下の様な粒度がありそう。先程の情報収集の結果を元に洗い出してみる。あるいはフルの状態のコードを元にコレを外せないか?アレが面倒なのだけど??とやっていくのも良い。ちょうどよく手軽だと感じた辺りで辞める。

  • 基礎的な部分だけで実行 (graphql-core-next)
  • ラッパーライブラリも含めて実行 (ariadne)
  • 実際に利用するときの形式だけで実行 (asgi appのhello world)
  • 実際に利用するときに形式に繋げた形で実行 (graphqlをasgi appとして実行)
  • E2E的に実行 (動作チェックはシェルスクリプトになる(今回は範囲外))

ariadneに関してはこの辺りを一度揃えておくと便利そう(最初から綺麗に考えられる事は少ないかもしれない。実際のところは試行錯誤の後の後付だったりする)。

requirements.txt

graphql-core>=3a
graphene
ariadne
uvicorn
async-asgi-testclient
pytest
pytest-asyncio

基礎的な部分だけで実行 (graphql-core-next)

同期版 00graphql-core-next.py

from graphql import GraphQLSchema, GraphQLObjectType, GraphQLField, GraphQLString

schema = GraphQLSchema(
    query=GraphQLObjectType(
        name="RootQueryType",
        fields={
            "hello": GraphQLField(GraphQLString, resolve=lambda obj, info: "world")
        },
    )
)


def main() -> None:
    from graphql import graphql_sync

    query = "{ hello }"

    print(graphql_sync(schema, query))


if __name__ == "__main__":
    main()

output

$ python 00graphql-core-next.py
ExecutionResult(data={'hello': 'world'}, errors=None)

非同期版

import asyncio
from graphql import GraphQLSchema, GraphQLObjectType, GraphQLField, GraphQLString


async def resolve_hello(obj, info):
    print("start resolve hello")
    await asyncio.sleep(1)
    print("finish resolve hello")
    return "world"


schema = GraphQLSchema(
    query=GraphQLObjectType(
        name="RootQueryType",
        fields={"hello": GraphQLField(GraphQLString, resolve=resolve_hello)},
    )
)


def main() -> None:
    from graphql import graphql

    async def run():
        query = "{ hello }"
        print(await graphql(schema, query))

    asyncio.run(run(), debug=True)


if __name__ == "__main__":
    main()

output

$ python 01graphql-core-next.py
start resolve hello
finish resolve hello
ExecutionResult(data={'hello': 'world'}, errors=None)

ラッパーライブラリも含めて実行 (ariadne)

02run-ariadne.py

import asyncio
import logging
from ariadne import ObjectType, gql, make_executable_schema, graphql


# Ariadne uses dedicated objects
query = ObjectType("Query")

# Map resolvers to fields in Query type using decorator syntax...
@query.field("hello")
def resolve_hello(_, info):
    request = info.context["request"]
    user_agent = request.get("HTTP_USER_AGENT", "guest")
    return "Hello, %s!" % user_agent


type_defs = gql(
    """
    type Query {
        hello: String!
    }
"""
)


logging.basicConfig(level=logging.DEBUG)
schema = make_executable_schema(type_defs, query)

query = {"query": "{ hello }"}
atask = graphql(
    schema, query, debug=True, context_value={"request": {"HTTP_USER_AGENT": "World"}}
)
print(asyncio.run(atask, debug=True))

output

$ python 02run-ariadne.py
DEBUG:asyncio:Using selector: EpollSelector
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>
(True, {'data': {'hello': 'Hello, World!'}})

実際に利用するときの形式だけで実行 (asgi appのhello world)

少し寄り道。asgi自体のテストの仕方も整理しておかないとめんどくさい。starletteなどのasgi用のフレームワークを使っている場合にはそのフレームワークが提供しているテストクライアントを使えば良いと言えば良いのだけれど。そうなると1ファイルの部分で説明したように依存が複雑になる。それは避けたいので別のライブラリを使う

The motivation behind this project is building a common testing library that doesn't depend on the web framework (Quart, Startlette, ...).

03test_asgi.py

import pytest
import async_asgi_testclient as aat


async def app(scope, receive, send):
    # assert scope["type"] == "http"
    await send(
        {
            "type": "http.response.start",
            "status": 200,
            "headers": [[b"content-type", b"text/plain"]],
        }
    )
    await send({"type": "http.response.body", "body": b"Hello, world!"})


@pytest.mark.asyncio
async def test_app():
    async with aat.TestClient(app) as client:
        resp = await client.get("/")
        assert resp.status_code == 200
        assert resp.text == "Hello, world!"

output

$ pytest -vv 03test_asgi.py
===================================== test session starts =====================================
platform linux -- Python 3.7.4, pytest-5.2.1, py-1.7.0, pluggy-0.12.0 -- $HOME/venvs/my/bin/python
cachedir: .pytest_cache
rootdir: $HOME/venvs/my/individual-sandbox/daily/20191021/example_ariadne
plugins: cov-2.6.1, asyncio-0.10.0
collected 1 item                                                                              

03test_asgi.py::test_app PASSED                                                         [100%]

====================================== 1 passed in 0.05s ======================================

実際に利用するときに形式に繋げた形で実行 (graphqlをasgi appとして実行)

繋げた形もテストコードの形式になっていると便利。

04test_ariadne.py

import pytest
from async_asgi_testclient import TestClient

from ariadne import ObjectType, gql, make_executable_schema
from ariadne.asgi import GraphQL

TestClient.__test__ = False  # prevent PytestCollectionWarning

# Ariadne uses dedicated objects
query = ObjectType("Query")

# Map resolvers to fields in Query type using decorator syntax...
@query.field("hello")
def resolve_hello(_, info):
    request = info.context["request"]
    user_agent = request.get("HTTP_USER_AGENT", "guest")
    return "Hello, %s!" % user_agent


type_defs = gql(
    """
    type Query {
        hello: String!
    }
"""
)

schema = make_executable_schema(type_defs, query)
# As standalone ASGI or WSGI app...
app = GraphQL(schema, debug=True)


@pytest.mark.asyncio
async def test_app():
    client = TestClient(app)

    resp = await client.post("/", json={"query": "{ hello }"})
    assert resp.status_code == 200
    assert resp.json() == {"data": {"hello": "Hello, guest!"}}

output

$ pytest -vv 04test_ariadne.py
===================================== test session starts =====================================
platform linux -- Python 3.7.4, pytest-5.2.1, py-1.7.0, pluggy-0.12.0 -- /home/nao/venvs/my/bin/python
cachedir: .pytest_cache
rootdir: /home/nao/venvs/my/individual-sandbox/daily/20191021/example_ariadne
plugins: cov-2.6.1, asyncio-0.10.0
collected 1 item                                                                              

04test_ariadne.py::test_app PASSED                                                      [100%]

====================================== 1 passed in 0.16s ======================================

おしまい。

手元で実行したいときのためのgist

参考

pythonとは直接関係ないけれどschema-firstのアプローチは嬉しいの嬉しくないのに対する1つの意見。

strawberry, ariadne, grapheneの比較。

API作成系のことについてのとても頑張っているメモ

pythonでmypyを騙してfieldにmetadataを付加したクラスを定義してみる

ちょっとだけ背景説明

なんでそんなことがしたくなったかというと、やっぱり定義を記述したらおしまいでいられる世界が理想なので。そして特定の領域(e.g. OpenAPI, GraphQL, Protcol Buffers, database定義, ...)に属さないような中立的な表現が欲しくなったので。

この記事は一言で言うならpythonのクラスの各フィールドにどうやってメタデータを持たせようか?ということに対するメモです。

どうしてpython?

現状よく触るのがgoとpythonなのですが、goは直和型が素直に記述できないので1

普及している言語の中ではTypeScriptの方がゆるふわな表現を良い感じに型パズルしやすいとは想うのですが、literal typetyped dictprotocolなどそこそこpythonでも悪くはない形で型チェックが可能になってきたかなーということが1つ。

あと、型情報が消えてしまわず直接値として参照できるという点もpythonのtype hintsは特殊で、実はこの部分が意外と便利に機能するのではないかという思いがあったりします。

そして複雑な型パズルがしたいというよりは至る所にメタデータをガチャガチャとくっつけたいという気持ちになったためです(reflect-metadata周りでtypescriptでも機能する気がしますし、最悪ASTを取り出してあれこれするという形でも良いので、やっぱり結局手に馴染んだものでのプロトタイプという意味合いが強い気がします)。

メタデータ無しの場合

例えば以下の様なpythonのコードがあったとします。

person.py

import typing as t


class Person:
    name: str
    age: int
    nickname: t.Optional[str]

このPersonというクラスから良い感じに情報を抜き出して使いまわそうと思っているのですがどうすれば良いでしょうか?仮の変換先としてOpenAPI Docのschemaの形式を利用することにします。

何もメタデータ無しの現状では以下の様に素直に変換できますね。

person.yaml

components:
  schemas:
    Person:
      properties:
        name:
          type: string
        age:
          type: integer
        nickname:
          type: string
      required:
      - name
      - age

メタデータありの場合

ここからが本題です。

クラスに対するメタデータ

クラスに対するメタデータの方法は以下の2つが考えられそうです。

  • デコレーター
  • なにか特定の値をクラス変数として仕込む

デコレーター

@metadata(x="xxx")
@metadata(y="yyy")
@metadata(z="zzz")
class Person:
    ...

なにか特定の値をクラス変数として仕込む

class Person:
    tablename = "people"

    ...

あるいはdescriptionなどはそのままdocstringが使えそうですね。

class Person:
    """this is person"""
    ...

クラスの方は対応方法が色々ありそうなのでどうにかなりそうな気がします。

クラスの各フィールドに対するメタデータ

今度はクラスの各フィールドに対する話です。

メソッドとデコレータ(上手くいかない)

クラスのときと同様にフィールドに対するデコレータもと考えると、メソッドにデコレーターでメタデータを付加すれば良いということになりますが。

class Person:

    @metadata(xxx="yyy")
    def name(self) -> str:
        """name of person""""
        ...

    ...

これはnameがメソッドであるというインターフェイスを前提としての記述になってしまうので嬉しくないです。以下の2つは違うので。

Person().name
Person().name()

そして可能ならDSLのためのベースとなるような記述が、他の部分の実際の実装に全く影響や前提を設けないというような形に持っていきたいです。

プロパティとデコレータ(上手くいかない)

それではということでプロパティにしてしまいましょう。duck-typingというやつです。

例えば以下の様な形です。

class Person:

    @property
    @metadata(xxx="yyy")
    def name(self) -> str:
        """name of person""""
        ...

    ...

ついでにドキュメントのためのプロパティということを表すためにmarker的な機能を付与したデコレータに変えていきたいところですが。これはまだまだpropertyをwrapしたgenericなデコレーターの型付けがサポートされていないっぽいので無理です。

class Person:

    @field(xxx="yyy")  # まだ無理(夢)
    def name(self) -> str:
        """name of person""""
        ...

    ...

加えてpropertyを直接使うとインターフェイスを固定してしまいますね。。。

class Person:
    @property
    def name(self) -> str:
        return ""


p = Person()
p.name = "foo"  # AttributeError: can't set attribute
print(p.name)

素直にインスタンス変数として使える形になっていて欲しいものです。

自前で定義したディスクリプターとデフォルト値(暫定的な回答)

そんなわけで以下をどうにかこうにか満たしたいわけです。

  • しっかりとmypyの型チェックが通るようにしたい
  • (なるべくmypyのpluginsなどには手を出したくない)
  • 利用方法に制限や前提を設けたくない(インターフェイスを固定したくない)

色々考えてみた結果の暫定的な回答は自前でディスクリプターを定義してそれをデフォルト値として使うということでした。

まず、なぜ自前でディスクリプターを定義すると良いのかと言うと、pyramidのreifyの妙と同じ話で、プロパティの機能の裏側にあるディスクリプターとインスタンス変数の優先順位の関係で、インスタンス変数への上書きを制限しなくなります。

class Person:
    @reify
    def name(self):
        return ""

p = Person()
print(p.name)  # => ""
p.name = "foo" # エラーにならない
print(p.name)  # => "foo"

また、インスタンス変数のデフォルト値をクラス変数として設定するのはmypyの型付けの流儀に沿ったものです2

# mypyドキュメントからの引用

class MyClass:
    # You can optionally declare instance variables in the class body
    attr: int
    # This is an instance variable with a default value
    charge_percent: int = 100

    # The "__init__" method doesn't return anything, so it gets return
    # type "None" just like any other method that doesn't return anything
    def __init__(self) -> None:
        ...

    # For instance methods, omit type for "self"
    def my_method(self, num: int, str1: str) -> str:
        return num * str1

dataclassesと似たような記述にすると言うと分かる人もいるかもしれません。ただdataclassesほど複雑ではないのでmypyの対応もpluginsという形での特別扱いは不要でした。

# documentから引用

@dataclass
class C:
    x: int
    y: int = field(repr=False)
    z: int = field(repr=False, default=10)
    t: int = 20

なぜディスクリプターが良いか

なぜディスクリプターが良いかというとメタデータにアクセスする方法を提供してくれるからです。ディスクリプター自体は通常の属性アクセスでは__get__()が呼ばれるのですが。

class Person:
    name : str = field("", x="xxx")  # fieldはディスクリプター

Person.name  # fieldの`__get__()` が呼ばれる

直接__dict__の中を覗くことでディスクリプタの実態にさわれます。

Person.__dict__["name"]  # <fieldのインスタンス>

そんなわけでディスクリプタ自身にメタデータをもたせておけば、外見上は属性アクセスを担保しつつフィールド毎にメタデータを保持することも可能になります。

# default値にアクセス
Person.name # => ""

# メタデータにアクセス
Person.__dict__["name"].metadata # => {"x": "xxx"}

実行時の感じとしては良さそうですね。

実際の定義

実際に動く実装を作って以下のように使えるようになることを目指します。

class Person:
    name: str
    age: int = 0


class WPerson(Person):
    nickname: t.Optional[str] = field(default=None, metadata=dict(doc="hmm"))

デフォルト値とメタデータを保持するディスクリプターの定義自体は以下の様な感じになります。が。

import typing as t

T = t.TypeVar("T")
MetaData = t.Optional[t.Dict[str, t.Any]]


class Field(t.Generic[T]):
    default: T
    metadata: t.Optional[MetaData]

    def __init__(self, default: T, *, metadata: t.Optional[MetaData] = None):
        self.default = default
        self.metadata = metadata

    def __get__(self, obj: object, type: t.Optional[t.Any] = None) -> T:
        return self.default

実際に上手くmypyの型チェックをすり抜けるために少しhack的なものが必要です。

def field(*, default: T, metadata: t.Optional[t.Dict[str, t.Any]] = None) -> T:
    return t.cast(T, Field(default, metadata=metadata))  # xxx: HACK

mypyにはあたかも以下が同じモノであるかのように見えるためのhackです。

class X:
    name : str
    name2 : str = ""
    name3 : str = field(default="", metadata={"x": "xxx"})

この様にやってあげると以下が実行もでき型チェックも通ります。

09typing.py

class Person:
    name: str
    age: int = field(default=0)


class WPerson(Person):
    nickname: t.Optional[str] = field(default=None, metadata=dict(doc="hmm"))


print(WPerson.nickname, WPerson.age)
print(get_metadata(WPerson, "nickname")) # metadataを取得する関数(gistに)

print("----------------------------------------")
# 各フィールドをiterateする関数(gistに)
for x in walk(WPerson):
    print(x)

if t.TYPE_CHECKING: # mypyのときだけ実行される場所
    reveal_type(WPerson.nickname)
    reveal_type(WPerson().nickname)


print("========================================")
wp = WPerson()
print(wp.nickname)
wp.nickname = "foo"
print(wp.nickname)

実行結果

$ python 09*.py 
None 0
{'doc': 'hmm'}
----------------------------------------
('name', <class 'str'>, None)
('age', <class 'int'>, None)
('nickname', typing.Union[str, NoneType], {'doc': 'hmm'})
========================================
None
foo

mypyの結果

$ $ mypy --strict --pretty 09*.py
09typing.py:54: note: Revealed type is 'Union[builtins.str, None]'
09typing.py:55: note: Revealed type is 'Union[builtins.str, None]'

reifyのようにデコレーターベースになっていると、メソッドのような形で記述してdocstringでのdescriptionをという夢も膨らみますが前述の通りでまだ無理です。

gist

自分で動作確認したい人のためのgist。

under construction

:construction: まだ説明する気は無いですが、この知見を使ってのパッケージを作成中です。 :construction:

github.com


  1. 厳密に言えばpythonのものもtagged unionではないものの、メンバーをクラスに絞ってsubtypingを考えなければ。。nominalなので。。クラス自身をタグにという扱いでとりあえず今回の要件は満たせる(クラスは値構築子)。

  2. 実行時の視点でみると、デフォルト値はself.__class__.<name>に保存され、__init__()などでの初期化は self.<name>に対するものなので大丈夫と言うような形。