Skip to main content

Python client side notification center.

Project description





PyNotiCenter is a multithreaded task queue and message notification center. It mainly provides in-process lightweight task scheduling management for client applications, that why we choose the multi-threaded mode. Although the multi-threaded mode is not a good practice most of the time in Python(In CPU-bound tasks it can't be genuinely parallel as the GIL exists). But in developing client applications, it will be one of the most important base components.

As it is based on a multi-threaded mode, multi-threaded development must pay attention to thread safety issues, which is inevitable. In order to reduce the complexity for developers, all interfaces of this library are thread safety. It is designed to be called in any thread to guarantee the same results as expected, even though this will cause some performance issues.

The library mainly provides 2 features.

  • Task queue: each task queue will run in a separate thread.
  • Message notification: the message notification scheduling is based on the built-in task queue.


pip install pynoticenter

Code Example

  • Post task to task queue.
def fn(*args: Any, **kwargs: Any):

def main():
    PyNotiCenter.default().post_task(fn, "hello world")
    PyNotiCenter.default().post_task_with_delay(5, False, fn, "hello", "world", "delay 5s")
[2022-09-03 20:54:23,698] {} INFO - TaskQueue[4408264928]: worker thread begin.
[2022-09-03 20:54:23,699] {} INFO - TaskQueue[4408264928]: Task queue terminate. wait: True
[2022-09-03 20:54:23,699] {} INFO - TaskQueue[4408264928]: waiting for tasks cleanup. tasks: 2
hello world
hello world delay 5s
[2022-09-03 20:54:28,721] {} INFO - TaskQueue[4408264928]: All tasks cleanup. wait time: 5.0
[2022-09-03 20:54:28,722] {} INFO - TaskQueue[4408264928]: waiting for thread exit.
[2022-09-03 20:54:28,722] {} INFO - TaskQueue[4408264928]: stop event run loop.
[2022-09-03 20:54:28,723] {} INFO - TaskQueue[4408264928]: worker thread end.
[2022-09-03 20:54:29,726] {} INFO - TaskQueue[4408264928]: thread exit. wait time: 1.0
# async func
async def async_fn(msg: str):
  await asyncio.sleep(1)
  print(f"msg: {msg}")

def fn(msg: str):
  print(f"msg: {msg}")

def main():
  PyNotiCenter.default().post_task(fn, "hello")
  PyNotiCenter.default().post_task(async_fn, "hello")
  • Notification
class A:
    def say_hello(self, who: str):
        print(f"{who}: hello")

def main():
    receiver = A()
    PyNotiCenter.default().add_observer("say_hello", receiver.say_hello, receiver)
    PyNotiCenter.default().notify_observers("say_hello", "baby")
  • Notification on specific thread, such as Gtk main thread.
def fn():

def switch_to_gtk_thread(fn: Callable, *args: Any, **kwargs) -> bool:
    return True

def main():
    queue = PyNotiCenter.default().create_task_queue(PyNotiOptions(queue='mytask'))
    queue.post_task(fn) # fn run in gtk thread

    # notification run on mytask queue.
    PyNotiCenter.default().add_observer("say_hello", fn, options=PyNotiOptions(queue="mytask"))

Post task with task id

def fn_with_task_id(task_id: str, msg: str):

def main():
    queue = PyNotiCenter.default().create_task_queue(PyNotiOptions(queue='mytask', fn_with_task_id=True))
    task_id = queue.post_task(fn_with_task_id, 'Hi')

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

pynoticenter-0.1.11.tar.gz (10.8 kB view hashes)

Uploaded source

Built Distribution

pynoticenter-0.1.11-py3-none-any.whl (12.3 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page