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 если контекст содержит значение для var; возвращает False в противном случае.

context[var]

Возвращает значение переменной контекста var. Если переменная не установлена в объекте контекста, генерируется исключение 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