contextvars — Context Variables

---

Цей модуль надає API для керування, зберігання та доступу до контекстно-локального стану. Клас ContextVar використовується для оголошення контекстних змінних і роботи з ними. Функцію copy_context() і клас Context слід використовувати для керування поточним контекстом в асинхронних структурах.

Context managers that have state should use Context Variables instead of threading.local() to prevent their state from bleeding to other code unexpectedly, when used in concurrent code.

Дивіться також PEP 567 для отримання додаткової інформації.

Нове в версії 3.7.

Контекстні змінні

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)

Виклик, щоб встановити нове значення для контекстної змінної в поточному контексті.

Обов’язковий аргумент значення — це нове значення для контекстної змінної.

Повертає об’єкт 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

A read-only property. Set to the value the variable had before the ContextVar.set() method call that created the token. It points to Token.MISSING if the variable was not set before the call.

MISSING

Об’єкт-маркер, який використовується Token.old_value.

Ручне керування контекстом

contextvars.copy_context()

Повертає копію поточного об’єкта Context.

Наступний фрагмент коду отримує копію поточного контексту та друкує всі змінні та їхні значення, встановлені в ньому:

ctx: Context = copy_context()
print(list(ctx.items()))

The function has an O(1) complexity, i.e. works equally fast for contexts with a few context variables and for contexts that have a lot of them.

class contextvars.Context

Відображення ContextVars на їхні значення.

Context() створює порожній контекст без значень у ньому. Щоб отримати копію поточного контексту, використовуйте функцію copy_context().

Every thread will have a different top-level Context object. This means that a ContextVar object behaves in a similar fashion to threading.local() when values are assigned in different threads.

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; інакше поверніть False.

context[var]

Повертає значення змінної var ContextVar. Якщо змінна не встановлена в об’єкті контексту, виникає KeyError.

get(var[, default])

Повертає значення для var, якщо var має значення в об’єкті контексту. Повернути за замовчуванням інакше. Якщо 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}\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