contextvars – Контекстные переменные¶contextvars – Context Variables

Контекстные переменные¶Context Variables

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')
# код, который использует 'var'; var.get() возвращает 'new value'.
var.reset(token)

# После вызова reset переменная снова не имеет значения, поэтому
# var.get() вызовет исключение LookupError.

class contextvars.Token¶

Объекты Token возвращаются методом ContextVar.set(). Их можно передать методу ContextVar.reset(), чтобы вернуть значение переменной к тому, которое было до соответствующего set.

var¶

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

old_value¶

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

MISSING¶

Объект-маркер, используемый Token.old_value.

Ручное управление контекстом¶Manual Context Management

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

context[var]

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

get(var[, default])¶

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

iter(context)

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

len(proxy)

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

keys()¶

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

values()¶

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

items()¶

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

Поддержка asyncio¶asyncio support

Контекстные переменные нативно поддерживаются в 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}\r\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(b'HTTP/1.1 200 OK\r\n')  # строка состояния
    writer.write(b'\r\n')  # заголовки
    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 или curl:
#     telnet 127.0.0.1 8081
#     curl 127.0.0.1:8081