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

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

class contextvars.ContextVar(name[, *, default])¶

Этот класс используется для объявления новой контекстной переменной, например:

var: ContextVar[int] = ContextVar('var', default=42)

Обязательный параметр name используется для целей интроспекции и отладки.

Необязательный параметр default, передаваемый только по ключу, возвращается ContextVar.get(), если в текущем контексте не найдено значение для переменной.

Важно: Контекстные переменные должны создаваться на верхнем уровне модуля, а не в замыканиях. Объекты Context содержат сильные ссылки на контекстные переменные, что препятствует их корректной сборке мусора.

ContextVar являются обобщёнными по типу содержащегося в них значения.

name¶

Имя переменной. Это свойство только для чтения.

Добавлено в версии 3.7.1.

get([default])¶

Возвращает значение для контекстной переменной в текущем контексте.

Если в текущем контексте нет значения для переменной, метод:

• возвращает значение аргумента default метода, если он был передан; или

• возвращает значение по умолчанию для контекстной переменной, если она была создана с ним; или

• возбуждает LookupError.

set(value)¶

Вызов для установки нового значения контекстной переменной в текущем контексте.

Обязательный аргумент value – это новое значение для контекстной переменной.

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

Для удобства объект токена можно использовать как менеджер контекста, чтобы избежать ручного вызова ContextVar.reset():

var = ContextVar('var', default='default value')

with var.set('new value'):
    assert var.get() == 'new value'

assert var.get() == 'default value'

Это краткая запись для:

var = ContextVar('var', default='default value')

token = var.set('new value')
try:
    assert var.get() == 'new value'
finally:
    var.reset(token)

assert var.get() == 'default value'

Добавлено в версии 3.14: Добавлена поддержка использования токенов в качестве менеджеров контекста.

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. Один токен не может сбросить контекстную переменную более одного раза.

Токены поддерживают протокол менеджера контекста для автоматического сброса контекстных переменных. См. ContextVar.set().

Токены являются обобщёнными относительно того же типа, что и ContextVar, который их создал.

Добавлено в версии 3.14: Добавлена поддержка использования в качестве менеджера контекста.

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. Текущий контекст – это объект Context на вершине стека текущего потока. Все объекты Context в стеках считаются вошедшими.

Вход в контекст, который можно выполнить вызовом его метода run(), делает контекст текущим, помещая его на вершину стека контекстов текущего потока.

Выход из текущего контекста, который можно выполнить возвратом из колбэка, переданного методу run(), восстанавливает текущий контекст до состояния, в котором он был до входа, путём удаления контекста с вершины стека контекстов.

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

Попытка войти в уже вошедший контекст, включая контексты, вошедшие в других потоках, вызывает RuntimeError.

После выхода из контекста его можно повторно войти (из любого потока).

Любые изменения значений ContextVar через метод ContextVar.set() записываются в текущий контекст. Метод ContextVar.get() возвращает значение, связанное с текущим контекстом. Выход из контекста эффективно отменяет любые изменения контекстных переменных, сделанные во время нахождения в контексте (при необходимости значения можно восстановить, повторно войдя в контекст).

Контекст реализует интерфейс collections.abc.Mapping.

run(callable, *args, **kwargs)¶

Входит в контекст, выполняет callable(*args, **kwargs), затем выходит из контекста. Возвращает возвращаемое значение вызываемого объекта или распространяет исключение, если оно произошло.

Пример:

import contextvars

var = contextvars.ContextVar('var')
var.set('spam')
print(var.get())  # 'spam'

ctx = contextvars.copy_context()

def main():
    # 'var' было установлено в 'spam' до
    # вызова 'copy_context()' и 'ctx.run(main)', поэтому:
    print(var.get())  # 'spam'
    print(ctx[var])  # 'spam'

    var.set('ham')

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

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

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

# Однако вне 'ctx' переменная 'var' по-прежнему равна 'spam':
print(var.get())  # 'spam'

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