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

Этот модуль предоставляет API для управления, хранения и доступа к контекстно-локальному состоянию. Класс ContextVar используется для объявления и работы с переменными контекста. Функция copy_context() и класс Context должны использоваться для управления текущим контекстом в асинхронных фреймворках.

Контекстные менеджеры, имеющие состояние, должны использовать переменные контекста вместо threading.local(), чтобы предотвратить утечку их состояния в другой код неожиданным образом при использовании в конкурентном коде.

См. также PEP 567 для дополнительной информации.

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

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(), создавшего токен.

Например:

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)

Выполняет 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. Если переменная не задана в объекте контекста, генерируется KeyError.

get(var[, default])

Возвращает значение для var, если var имеет значение в объекте контекста. В противном случае возвращает default. Если default не задан, возвращает None.

iter(context)

Возвращает итератор по переменным, хранящимся в объекте контекста.

len(proxy)

Возвращает количество установленных переменных в объекте контекста.

keys()

Возвращает список всех переменных в объекте контекста.

values()

Возвращает список значений всех переменных в объекте контекста.

items()

Возвращает список пар вида (переменная, значение) для всех переменных в объекте контекста.

Поддержка 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