Factory for contextvar default value

Question

Setting a dictionary as ContextVar default:

var: ContextVar[dict] = ContextVar('var', default={})

...kinda works, as the dictionary will be available as default, but it always references the same instance, instead of generating a new one for each context.

Do contextvars somehow support factories (for dicts, lists, and alike), as in:

var: ContextVar[dict] = ContextVar('var', default=dict)
var: ContextVar[dict] = ContextVar('var', default=lambda: dict())

Or do I just have to do it manually:

var: ContextVar[Optional[dict]] = ContextVar('var', default=None)

...

if not var.get():
    var.set({})

Answer 1

Apparently ContextVars designed choice was in the direction of providing the low level, barebones, functionality over ease of use.

There is no easy way to get a context-aware namespace, as you intend to do by having the dictionary as default. And also, no option for the default value to use a factory rather than a single object.

The only way to overcome that is to write a class that provides a higher level interface on top of contextvars (or other context-separating mechanism).

I am just working on such a package, although I made no release yet - y main goal is to have a class that act as a free-to-use namespace, just like threading.Local instances. (There is also one class using the mapping interface) - if I get more people using and providing some feedback, I could come faster to a finished form:

https://github.com/jsbueno/extracontext

Answer 2

I recommend to use Google's wrapper around contextvar: https://github.com/google/etils/blob/main/etils/edc/README.md#wrap-fields-around-contextvar for a dataclass-like API:

Any dataclass is supported:

• To make a field context-dependent: annotate the field with edc.ContextVar[T].
• To have a factory as default value, use: dataclasses.field(default_factory=)

from etils import edc

@edc.dataclass
@dataclasses.dataclass
class Context:
  thread_id: edc.ContextVar[int] = dataclasses.field(default_factory=threading.get_native_id)

  # Local stack: each thread will use its own instance of the stack
  stack: edc.ContextVar[list[str]] = dataclasses.field(default_factory=list)


# Global context object
context = Context(thread_id=0)

Each threads/task will have it's own version of the field:

def worker():
  # Inside each thread, the worker use its own context
  assert context.thread_id != 0
  context.stack.append(1)
  time.sleep(1)
  assert len(context.stack) == 1  # Other workers do not modify the local stack

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
  for _ in range(10):
    executor.submit(worker)