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(), который создал токен.

Например:

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

Контекст реализует интерфейс 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 если context имеет значение для var; в противном случае возвращает False.

context[var]

Возвращает значение переменной контекста var. Если переменная не установлена в объекте контекста, генерируется исключение 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():
    # 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}\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(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())

# To test it you can use telnet:
#     telnet 127.0.0.1 8081