contextvars --- コンテキスト変数

---

このモジュールは、コンテキストローカルな状態を管理し、保持し、アクセスするための API を提供します。 ContextVar クラスは コンテキスト変数 を宣言し、取り扱うために使われます。 非同期フレームワークで現時点のコンテキストを管理するのには、 copy_context() 関数と Context クラスを使うべきです。

状態を持っているコンテキストマネージャは threading.local() ではなくコンテキスト変数を使い、並行処理のコードから状態が意図せず他のコードへ漏れ出すのを避けるべきです。

より詳しくは PEP 567 を参照をしてください。

バージョン 3.7 で追加.

コンテキスト変数

class contextvars.ContextVar(name[, *, default])

このクラスは新しいコンテキスト変数を宣言するのに使われます。例えば、次の通りです:

var: ContextVar[int] = ContextVar('var', default=42)

必須のパラメータの name は内観やデバッグの目的で使われます。

オプションのキーワード専用引数 default は、現在のコンテキストにその変数の値が見付からなかったときに ContextVar.get() から返されます。

重要: コンテキスト変数は、モジュールのトップレベルで生成する必要があり、クロージャの中で作成すべきではありません。Context オブジェクトはコンテキスト変数への強参照を持っており、コンテキスト変数がガーベジコレクトされるのを防ぎます。

name

変数の名前。読み出し専用のプロパティです。

バージョン 3.7.1 で追加.

get([default])

現在のコンテキストのコンテキスト変数の値を返します。

現在のコンテキストのコンテキスト変数に値がなければ、メソッドは:

• メソッドの default 引数に値が指定されていればその値を返します。さもなければ

• コンテキスト変数が生成された時にデフォルト値が渡されていれば、その値を返します。さもなければ

• LookupError を送出します。

set(value)

現在のコンテキストのコンテキスト変数に新しい値を設定する際に呼び出します。

value は必須の引数で、コンテキスト変数の新しい値を指定します。

Token オブジェクトを返します。このオブジェクトを ContextVar.reset() メソッドに渡すことで、以前の値に戻すことができます。

reset(token)

コンテキスト変数の値を、 token を生成した ContextVar.set() が呼び出される前の値にリセットします。

例えば:

var = ContextVar('var')

token = var.set('new value')
# code that uses 'var'; var.get() returns 'new value'.
var.reset(token)

# After the reset call the var has no value again, so
# var.get() would raise a LookupError.

class contextvars.Token

Token オブジェクトは、ContextVar.set() メソッドによって返されるオブジェクトです。このオブジェクトを ContextVar.reset() メソッドに渡すことで、対応する set を呼び出す前のコンテキスト変数の値に戻せます。

Token.var

読み出し専用のプロパティです。トークンを生成した ContextVar オブジェクトを指します。

Token.old_value

読み出し専用のプロパティです。このトークンを返した ContextVar.set() メソッドの呼び出し前に設定されていた値を返します。もし呼び出しの前に値が設定されていなければ Token.MISSING を返します。

Token.MISSING

Token.old_value で利用されるマーカーオブジェクトです。

マニュアルでのコンテキスト管理

contextvars.copy_context()

現在の Context オブジェクトのコピーを返します。

次のスニペットは、現在のコンテキストのコピーを取得し、コンテキストに設定されているすべての変数とその値を表示します:

ctx: Context = copy_context()
print(list(ctx.items()))

この関数の複雑性はO(1) です。つまり、少数のコンテキスト変数を持つコンテキストと多くの変数を持つコンテキストで同程度の速度で動作します。

class contextvars.Context

ContextVars とその値の写像。

Context() は、値を持たない空のコンテキストを生成します。現在のコンテキストのコピーを得るには、copy_context() 関数を利用します。

Context は、 collections.abc.Mapping インタフェースを実装します。

run(callable, *args, **kwargs)

callable(*args, **kwargs) を run メソッドが呼ばれたコンテキストオブジェクトの中で実行します。実行した結果を返すか、例外が発生した場合はその例外を伝播します。

callable が行ったコンテキスト変数へのいかなる変更も、コンテキストオブジェクトに格納されます:

var = ContextVar('var')
var.set('spam')

def main():
    # 'var' was set to 'spam' before
    # calling 'copy_context()' and 'ctx.run(main)', so:
    # var.get() == ctx[var] == 'spam'

    var.set('ham')

    # Now, after setting 'var' to 'ham':
    # var.get() == ctx[var] == 'ham'

ctx = copy_context()

# Any changes that the 'main' function makes to 'var'
# will be contained in 'ctx'.
ctx.run(main)

# The 'main()' function was run in the 'ctx' context,
# so changes to 'var' are contained in it:
# ctx[var] == 'ham'

# However, outside of 'ctx', 'var' is still set to 'spam':
# var.get() == 'spam'

2つ以上の OS スレッドから同一のコンテキストオブジェクトを呼び出すか、再帰的に呼び出したとき、メソッドは RuntimeError を送出します。

copy()

コンテキストオブジェクトの浅いコピーを返します。

var in context

context に var の値が設定されていた場合 True を返します; そうでない場合は False を返します。

context[var]

ContextVar var の値を返します。コンテキストオブジェクト内で変数が設定されていない場合は、KeyError を送出します。

get(var[, default])

var がコンテキストオブジェクトの中に値を持てば、その値を返します。さもなければ、default を返します。default を指定していなければ、None を返します。

iter(context)

コンテキストオブジェクトに格納されている変数群のイテレータを返します。

len(proxy)

コンテキストオブジェクトに格納されている変数の数を返します。

keys()

コンテキストオブジェクト中のすべての変数のリストを返します。

values()

コンテキストオブジェクト中のすべての変数の値のリストを返します。

items()

コンテキストオブジェクト中のすべての変数について、変数とその値からなる2要素のタプルのリストを返します。

asyncio サポート

コンテキスト変数は、追加の設定なしに asyncio をサポートします。例えば、次の単純なechoサーバーは、クライアントを扱う Task の中でリモートクライアントのアドレスが利用できるように、コンテキスト変数を利用します:

import asyncio
import contextvars

client_addr_var = contextvars.ContextVar('client_addr')

def render_goodbye():
    # The address of the currently handled client can be accessed
    # without passing it explicitly to this function.

    client_addr = client_addr_var.get()
    return f'Good bye, client @ {client_addr}\n'.encode()

async def handle_request(reader, writer):
    addr = writer.transport.get_extra_info('socket').getpeername()
    client_addr_var.set(addr)

    # In any code that we call is now possible to get
    # client's address by calling 'client_addr_var.get()'.

    while True:
        line = await reader.readline()
        print(line)
        if not line.strip():
            break
        writer.write(line)

    writer.write(render_goodbye())
    writer.close()

async def main():
    srv = await asyncio.start_server(
        handle_request, '127.0.0.1', 8081)

    async with srv:
        await srv.serve_forever()

asyncio.run(main())

# To test it you can use telnet:
#     telnet 127.0.0.1 8081