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() возвращает 'новое значение'.
var.reset(token)

# После вызова reset переменная снова не имеет значения, поэтому
# 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 при вызове одного и того же объекта контекста из более чем одного потока ОС или при рекурсивном вызове.

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()

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

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