contextvars —コンテキスト変数—Pythonドキュメント

contextvars —コンテキスト変数

---

このモジュールは、コンテキストローカル状態を管理、保存、およびアクセスするためのAPIを提供します。 ContextVar クラスは、 Context Variables を宣言して操作するために使用されます。 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 引数は、コンテキスト変数の新しい値です。

ContextVar.reset（）メソッドを介して変数を以前の値に復元するために使用できる Token オブジェクトを返します。

reset(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（）関数を使用します。

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

run(callable, *args, **kwargs)

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

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'

このメソッドは、複数のOSスレッドから同じコンテキストオブジェクトで呼び出された場合、または再帰的に呼び出された場合に RuntimeError を発生させます。

copy()

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

var in context

context に var の値が設定されている場合は、Trueを返します。 それ以外の場合はFalseを返します。

context[var]

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

get(var[, default])

var の値がコンテキストオブジェクトにある場合は、 var の値を返します。 それ以外の場合は、 default を返します。 default が指定されていない場合は、Noneを返します。

iter(context)

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

len(proxy)

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

keys()

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

values()

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

items()

コンテキストオブジェクト内のすべての変数とその値を含む2タプルのリストを返します。

asyncioのサポート

コンテキスト変数は asyncio でネイティブにサポートされており、追加の構成なしですぐに使用できます。 たとえば、これは単純なエコーサーバーであり、コンテキスト変数を使用して、リモートクライアントのアドレスをそのクライアントを処理するタスクで使用できるようにします。

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