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.

Filename, size & hash SHA256 hash help File type Python version Upload date
metapensiero.asyncio.tasklocal-0.2.tar.gz (4.7 kB) Copy SHA256 hash SHA256 Source None

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page