contextvars — Переменные контекста

Переменные контекста

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(), создавшего 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.

Один и тот же token нельзя использовать дважды.

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.

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

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

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

run(callable, *args, **kwargs)

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

Пример:

import contextvars

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

ctx = contextvars.copy_context()

def main():
    # 'var' was set to 'spam' before
    # calling 'copy_context()' and 'ctx.run(main)', so:
    print(var.get())  # 'spam'
    print(ctx[var])  # 'spam'

    var.set('ham')

    # Now, after setting 'var' to 'ham':
    print(var.get())  # 'ham'
    print(ctx[var])  # 'ham'

# 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:
print(ctx[var])  # 'ham'

# However, outside of 'ctx', 'var' is still set to 'spam':
print(var.get())  # 'spam'

copy()

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

var in context

Возвращает True, если в context задано значение для var; в противном случае возвращает False.

context[var]

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

get(var[, default])

Возвращает значение для 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():
    # 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}\r\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(b'HTTP/1.1 200 OK\r\n')  # status line
    writer.write(b'\r\n')  # headers
    writer.write(render_goodbye())  # body
    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 or curl:
#     telnet 127.0.0.1 8081
#     curl 127.0.0.1:8081