contextvars — Контекстные переменные

Контекстные переменные

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')
# 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.

class contextvars.Token

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

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 верхнего уровня. Это означает, что объект ContextVar ведёт себя аналогично threading.local(), когда значения присваиваются в разных потоках.

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

run(callable, *args, **kwargs)

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

Все изменения контекстных переменных, внесённые объектом callable, будут сохранены в объекте контекста:

var = ContextVar('var')
var.set('spam')

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

    var.set('ham')

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

ctx = copy_context()

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

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

Метод вызывает исключение RuntimeError, если он вызван для одного и того же объекта контекста из нескольких потоков ОС или рекурсивно.

copy()

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

var in context

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

context[var]

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

get(var[, default])

Возвращает значение для 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():
    # 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