contextvars --- متغیرهای زمینه

---

این ماژول APIهایی را برای مدیریت، ذخیره و دسترسی به وضعیت محلی زمینه فراهم می‌کند. از کلاس ContextVar برای اعلام و کار با متغیرهای زمینه استفاده می‌شود. برای مدیریت زمینه جاری در چارچوب‌های ناهمگام باید از تابع copy_context() و کلاس Context استفاده شود.

مدیران زمینه‌ای که وضعیت دارند، باید به‌جای threading.local() از متغیرهای زمینه استفاده کنند تا هنگام استفاده در کد هم‌روند، از نشت وضعیتشان به سایر کدها به‌طور غیرمنتظره جلوگیری شود.

همچنین برای جزئیات بیشتر، PEP 567 را ببینید.

اضافه شده در نسخه‌ی 3.7.

متغیرهای زمینه

class contextvars.ContextVar(name[, *, default])

این کلاس برای اعلام یک متغیر زمینه (Context Variable) جدید به کار می‌رود، برای مثال:

var: ContextVar[int] = ContextVar('var', default=42)

پارامتر الزامی name برای درون‌نگری و اهداف اشکال‌زدایی استفاده می‌شود.

پارامتر اختیاری فقط کلیدواژه‌ای default توسط ContextVar.get() برگردانده می‌شود، هنگامی که هیچ مقداری برای متغیر در زمینه جاری یافت نشود.

مهم: متغیرهای زمینه باید در سطح بالای ماژول ایجاد شوند و هرگز در بسته‌ها (closures) ایجاد نشوند. اشیای 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.

نمی‌توان از همان توکن دو بار استفاده کرد.

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

اگر context دارای مقداری برای var تنظیم‌شده باشد، True را برمی‌گرداند؛ در غیر این صورت False را برمی‌گرداند.

context[var]

مقدار متغیر var از کلاس ContextVar را برمی‌گرداند. اگر متغیر در شیء زمینه تنظیم‌نشده باشد، استثنای KeyError پرتاب می‌شود.

get(var[, default])

اگر var در شیء زمینه مقدار داشته باشد، مقدار var را برمی‌گرداند. در غیر این صورت، default را برمی‌گرداند. اگر default داده نشده باشد، None را برمی‌گرداند.

iter(context)

یک پیمایش‌گر بر روی متغیرهای ذخیره‌شده در شیء زمینه برمی‌گرداند.

len(proxy)

تعداد متغیرهای تنظیم‌شده در شیء زمینه را برمی‌گرداند.

keys()

فهرستی از همه متغیرهای شیء زمینه را برمی‌گرداند.

values()

فهرستی از مقادیر تمام متغیرها در شیء زمینه برمی‌گرداند.

items()

فهرستی از تاپل‌های دوتایی را برمی‌گرداند که شامل تمام متغیرها و مقادیر آن‌ها در شیء زمینه است.

پشتیبانی از asyncio

متغیرهای زمینه به‌صورت بومی در asyncio پشتیبانی می‌شوند و بدون هیچ‌گونه پیکربندی اضافی آماده استفاده هستند. برای مثال، در اینجا یک سرور اکو ساده آمده است که از یک متغیر زمینه استفاده می‌کند تا نشانی یک کلاینت دوردست را در Taskای که آن کلاینت را مدیریت می‌کند در دسترس قرار دهد:

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