Welcome, guest | Sign In | My Account | Store | Cart

Notice! PyPM is being replaced with the ActiveState Platform, which enhances PyPM’s build and deploy capabilities. Create your free Platform account to download ActivePython or customize Python with the packages you require and get automatic updates.

pypm install gocept.async

How to install gocept.async

  1. Download and install ActivePython
  2. Open Command Prompt
  3. Type pypm install gocept.async
 Python 2.7Python 3.2Python 3.3
Windows (32-bit)
0.3.3 Available View build log
0.3.2 Available View build log
Windows (64-bit)
0.3.3 Available View build log
0.3.2 Available View build log
Mac OS X (10.5+)
0.3.3 Available View build log
0.3.2 Available View build log
Linux (32-bit)
0.3.3 Available View build log
0.3.2 Available View build log
Linux (64-bit)
0.3.3 Available View build log
0.3.2 Available View build log
ZPL 2.1
Lastest release
version 0.3.3 on Apr 5th, 2011

gocept.async provides asynchronous function calls using a decorator. It facilitates lovely.remotetask to actually do things asynchronously.


It is often desirable to process computations asynchronously. Until there was lovely.remotetask this was not so easy to achieve in a Zope 3 application due to transaction integration issues.

gocept.async makes the task even easier:

>>> import gocept.async
>>> @gocept.async.function(service='events')
... def heavy_computing(a, b):
...     print "Computing", a, "+", b, "=", a + b

The decorator gocept.async.function takes exactly one argument, the name of a lovely.remotetask.interfaces.ITaskService utility. Note that gocept.async does not define any task service by itself.

Test Setup

Note that the decorated function must have an importable module to be usable:

>>> import gocept.async.tests
>>> heavy_computing.undecorated.__module__ = 'gocept.async.tests'
>>> gocept.async.tests.heavy_computing = heavy_computing

We defined task-service called events in this test:

>>> import zope.component
>>> import lovely.remotetask
>>> import lovely.remotetask.interfaces
>>> import lovely.remotetask.processor
>>> sm = zope.component.getSiteManager()
>>> getRootFolder()['tasks'] = tasks = lovely.remotetask.TaskService()
>>> tasks.processorFactory = lovely.remotetask.processor.MultiProcessor
>>> tasks.processorArguments = {'maxThreads': 1}
>>> sm.registerUtility(
...     tasks, lovely.remotetask.interfaces.ITaskService, name='events')


When the decorated function is called it returns nothing:

>>> heavy_computing(2, 7)

When we start the processing of the task service, the function is called:

>>> gocept.async.tests.process()
Computing 2 + 7 = 9

When the function is called while a user is logged in, the function will be called as that user (Note that it might be necessary to manually create security proxies to enable security in the async function.):

>>> @gocept.async.function('events')
... def who_am_i():
...     print gocept.async.task.TaskDescription.get_principal()
>>> who_am_i.undecorated.__module__ = 'gocept.async.tests'
>>> gocept.async.tests.who_am_i = who_am_i
>>> who_am_i()
>>> gocept.async.tests.process()

Now login:

>>> gocept.async.tests.login('zope.user')
>>> who_am_i()
>>> gocept.async.tests.process()
>>> gocept.async.tests.logout()

If an async function is called while the process is already async'ed the function is called immediately:

>>> @gocept.async.function(service='events')
... def call_another():
...     print "Before"
...     heavy_computing(1, 2)
...     print "After"
>>> call_another.undecorated.__module__ = 'gocept.async.tests'
>>> gocept.async.tests.call_another = call_another
>>> call_another()
>>> gocept.async.tests.process()
Computing 1 + 2 = 3

There is a helper to test if currenlty async is in progress:

>>> gocept.async.is_async()
>>> @gocept.async.function(service='events')
... def is_async_test():
...     print gocept.async.is_async()
>>> is_async_test.undecorated.__module__ = 'gocept.async.tests'
>>> gocept.async.tests.is_async_test = is_async_test
>>> is_async_test()
>>> gocept.async.tests.process()


>>> sm.registerUtility(
...     tasks, lovely.remotetask.interfaces.ITaskService, name='events')
>>> del gocept.async.tests.heavy_computing
>>> del gocept.async.tests.who_am_i
>>> del gocept.async.tests.call_another
>>> del gocept.async.tests.is_async_test


0.3.3 (2011-04-05)
  • Replace deprecated zope.testing.doctest with stdlib's doctest.
0.3.2 (2010-04-14)
  • Fix conflict error logging.
0.3.1 (2009-09-02)
  • Extracted a function to determine wether a function is run asynchronously or not.
0.3 (2009-07-31)
  • Remember the site that was active when the async function was defined and restore it when it is run.
0.2 (2009-04-16)
  • Made it even less likely that tests will raise ConflictErrors.
  • Fixed tests in README.
0.1.1 (2009-02-11)
  • Made it less likely that tests will raise ConflictErrors.
0.1 (2009-02-11)
  • first internal release.

Subscribe to package updates

Last updated Apr 5th, 2011

Download Stats

Last month:5

What does the lock icon mean?

Builds marked with a lock icon are only available via PyPM to users with a current ActivePython Business Edition subscription.

Need custom builds or support?

ActivePython Enterprise Edition guarantees priority access to technical support, indemnification, expert consulting and quality-assured language builds.

Plan on re-distributing ActivePython?

Get re-distribution rights and eliminate legal risks with ActivePython OEM Edition.