contextvars — Переменные контекста

Переменные контекста

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)

Восстанавливает переменную контекста до значения, которое у неё было до вызова ContextVar.set(), который создал token.

Например:

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.

var

Свойство только для чтения. Указывает на объект ContextVar, который создал токен.

old_value

Свойство только для чтения. Устанавливается в значение, которое имела переменная до вызова метода ContextVar.set(), который создал токен. Указывает на Token.MISSING, если переменная не была установлена до вызова.

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. Это означает, что объект ContextVar ведет себя аналогично объекту threading.local(), когда значения присваиваются в разных потоках.

Контекст реализует интерфейс 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'

Метод вызывает RuntimeError, когда вызывается на одном и том же объекте контекста из более чем одного потока ОС или при рекурсивном вызове.

copy()

Возвращает неглубокую копию объекта контекста.

var in context

Возвращает True если context имеет значение для var; возвращает 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