Skip to main content

An asyncio's Task-local variable container

Project description

author:Alberto Berti
license:GNU General Public License version 3 or later

An asyncio’s Task-local variable container


A drop-in replacement for threading.local object that stores per-task variables (instead of per-thread). You can use it like this:

import asyncio
from metapensiero.asyncio.tasklocal import Local

mylocal = Local()

def task_a():
    mylocal.a_var = 'foo'
    yield from asyncio.sleep(0.001) # switches to task_b
    assert mylocal.a_var == 'foo'

def task_b():
    mylocal.a_var = 'bar'
    yield from asyncio.sleep(0.001) # switches back to task_a
    assert mylocal.a_var == 'bar'

loop = asyncio.get_event_loop()
a = asyncio.Task(task_a())
b = asyncio.Task(task_b())
loop.run_until_complete(asyncio.wait((a, b)))

# this prints out:
# foo
# bar

Really the kind of object to be tracked as “current” is configurable by passing-in a different Discriminator instance to the Local class.

By default it will use the TaskDiscriminator supplied with the package. It will use standard asyncio’s loop detection techniques (by delegating it to asyncio.Task.current_task()) but if you want to track tasks of a particular event loop, you can supply your own TaskDiscriminator instance to the Local instance initialization, like this:

from metapensiero.asyncio.tasklocal import Local, TaskDiscriminator

mylocal = Local(discriminator=TaskDiscriminator(loop=my_loop))

You can also track another set of object by subclassing the BaseDiscriminator class:

from metapensiero.asyncio.tasklocal import BaseDiscriminator

class ASampleSessionDiscriminator(BaseDiscriminator):

    def current(self):
        return current_session()

mylocal = Local(discriminator=ASampleDiscriminator())

You have to implement at least the current method.

Like threading.local class, this class supports subclassing, in which case it will re-execute the __init__ method for each one of the tracked objects:

call_counter = 0

class CustomLocal(Local):

    a_value = None

    def __init__(self, a_value):
        self.a_value = a_value
        nonlocal call_counter
        call_counter += 1

mylocal = CustomLocal('foo')

Here mylocal.a_value will be initialized to foo for every tracked object (asyncio’s tasks by default). Here call_counter will count the number of every tracked object in which the mylocal object has been accessed.


To run the tests you should run the following at the package root:

python test

Build status


0.1 (2015-12-14)
  • First release complete with tests
0.2 (2015-12-26)
  • Fix CHANGES and packaging

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

metapensiero.asyncio.tasklocal-0.2.tar.gz (4.7 kB view hashes)

Uploaded source

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor NVIDIA NVIDIA PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page