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

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

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

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

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

Любые изменения в любых контекстных переменных, которые вносит callable, будут содержаться в объекте context:

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

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

var in context

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

context[var]

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

get(var[, default])¶

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

iter(context)

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

len(proxy)

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

keys()¶

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

values()¶

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

items()¶

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

поддержка 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