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)

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

Например:

var = ContextVar('var')

token = var.set('new value')
# код, который использует 'var'; var.get() возвращает 'new value'.
var.reset(token)

# После вызова сброса переменная снова не имеет значения, поэтому
# var.get() поднимет 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' был установлен как 'spam' перед тем
    # вызывать 'copy_context()' и 'ctx.run(main)', таким образом:
    # var.get() == ctx[var] == 'spam'

    var.set('ham')

    # Теперь, после установки 'var' в 'ham':
    # var.get() == ctx[var] == 'ham'

ctx = copy_context()

# Любые изменения, которые функция 'main' вносит в 'var'
# будет содержаться в 'ctx'.
ctx.run(main)

# Функция 'main()' была запущена в контексте 'ctx',
# поэтому изменения в 'var' содержатся в нем:
# ctx[var] == 'ham'

# Тем не менее, вне 'ctx', 'var' по-прежнему установлен в 'spam':
# var.get() == 'spam'

Метод поднимает RuntimeError, когда его называют на том же объекте контекста больше чем от одной потока OS, или когда его называют рекурсивно.

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():
    # К адресу текущего обрабатываемого клиента можно обращаться без явной передачи
    # его этой функции.

    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)

    # Любой код, который мы вызываем, теперь возможет получить адрес клиента,
    # вызвав '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())

# Чтобы проверить воспользуемся telnet:
#     telnet 127.0.0.1 8081