Skip to content

Instantly share code, notes, and snippets.

@diosmosis
Created August 8, 2011 18:46
Show Gist options
  • Save diosmosis/1132418 to your computer and use it in GitHub Desktop.
Save diosmosis/1132418 to your computer and use it in GitHub Desktop.
Asynchronous function calling with python and GTK.
import threading
from gi.repository import GObject
# calls f on another thread
def async_call(f, on_done):
"""
Starts a new thread that calls f and schedules on_done to be run (on the main
thread) when GTK is not busy.
Args:
f (function): the function to call asynchronously. No arguments are passed
to it. f should not use any resources used by the main thread,
at least not without locking.
on_done (function): the function that is called when f completes. It is
passed f's result as the first argument and whatever
was thrown (if anything) as the second. on_done is
called on the main thread, so it can access resources
on the main thread.
Returns:
Nothing.
Raises:
Nothing.
"""
if not on_done:
on_done = lambda r, e: None
def do_call():
result = None
error = None
try:
result = f()
except Exception, err:
error = err
GObject.idle_add(lambda: on_done(result, error))
thread = threading.Thread(target = do_call)
thread.start()
# free function decorator
def async_function(on_done = None):
"""
A decorator that can be used on free functions so they will always be called
asynchronously. The decorated function should not use any resources shared
by the main thread.
Example:
@async_function(on_done = do_whatever_done)
def do_whatever(look, at, all, the, pretty, args):
# ...
Args:
on_done (function): the function that is called when the decorated function
completes. If omitted or set to None this will default
to a no-op. This function will be called on the main
thread.
on_done is called with the decorated function's result
and any raised exception.
Returns:
A wrapper function that calls the decorated function on a new thread.
Raises:
Nothing.
"""
def wrapper(f):
def run(*args, **kwargs):
async_call(lambda: f(*args, **kwargs), on_done)
return run
return wrapper
# method decorator
def async_method(on_done = None):
"""
A decorator that can be used on class methods so they will always be called
asynchronously. The decorated function should not use any resources shared
by the main thread.
Example:
@async_method(on_done = lambda self, result, error: self.on_whatever_done(result, error))
def do_whatever(self, look, at, all, the, pretty, args):
# ...
Args:
on_done (function): the function that is called when the decorated function
completes. If omitted or set to None this will default
to a no-op. This function will be called on the main
thread.
on_done is called with the class instance used, the
decorated function's result and any raised exception.
Returns:
A wrapper function that calls the decorated function on a new thread.
Raises:
Nothing.
"""
if not on_done:
on_done = lambda s, r, e: None
def wrapper(f):
def run(self, *args, **kwargs):
async_call(lambda: f(self, *args, **kwargs), lambda r, e: on_done(self, r, e))
return run
return wrapper
@JasonLG1979
Copy link

Thanks. I found this looking for how to do async calls in PyGObject. I have adapted it for my needs:
https://gist.github.com/JasonLG1979/e3ab511a60b64f5cdaa0a643d3f3540f

@tallero
Copy link

tallero commented May 9, 2021

Unfortunately when I use this I logging always screw up.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment