API Documentation¶

How to import package¶

   from pynoticenter import PyNotiCenter

   def fn():
      pass

   def main():
      PyNotiCenter.default().post_task(fn)

PyNotiCenter¶

class PyNotiCenterInterface¶

Bases: ABC

PyNotiCenter Interface

abstract add_observer(name: str, fn: Callable, receiver: Optional[Any] = None, *, options: Optional[PyNotiOptions] = None)¶

Add observer to PyNotiCenter

Parameters:

name (str) – notification name
fn (Callable) – callback function
receiver (Any) – receiver object
options (PyNotiOptions) – options

abstract cancel_task(task_id)¶

cancel task from default task queue with task id

Parameters:: task_id (str) – task id

abstract cancel_task_with_queue_name(queue_name: str, task_id: str)¶

cancel task from named task queue with task id

Parameters:

queue_name (str) – task queue name
task_id (str) – task id

abstract create_task_queue(options: PyNotiOptions) → PyNotiTaskQueue¶

create task queue with name

Parameters:: options (PyNotiOptions) – task queue options
Returns:: task queue
Return type:: PyNotiTaskQueue

abstract get_default_task_queue() → PyNotiTaskQueue¶

return default task queue

Returns:: default task queue
Return type:: PyNotiTaskQueue

abstract get_task_queue(queue_name: str) → PyNotiTaskQueue¶

return named task queue

Parameters:: queue_name (str) – task queue name
Returns:: task queue
Return type:: PyNotiTaskQueue

abstract notify_observers(name: str, *args: Any, **kwargs: Any)¶

Notify observers

Parameters:

name (str) – notification name
*args (Any) – args
**kwargs (Any) – kwargs

abstract post_task(fn: Callable, *args: Any, **kwargs: Any) → str¶

post task to default task queue.

Parameters:

fn (Callable) – callback function
*args (Any) – args
**kwargs (Any) – kwargs

Returns:

return task id

Return type:

str

abstract post_task_to_task_queue(queue_name: str, fn: Callable, *args: Any, **kwargs: Any) → str¶

post task to named task queue.

Parameters:

fn (Callable) – callback function
queue_name (str) – queue name, create from create_task_queue
*args (Any) – args
**kwargs (Any) – kwargs

Returns:

return task id

Return type:

str

abstract post_task_with_delay(delay: float, fn: Callable, *args: Any, **kwargs: Any) → str¶

post task to default task queue with delay.

Parameters:

fn (Callable) – callback function
delay (float) – delay time in seconds.
*args (Any) – args
**kwargs (Any) – kwargs

Returns:

return task id

Return type:

str

abstract release_task_queue(queue_name: str, wait: bool)¶

release task queue

Parameters:

queue_name (str) – task queue name
wait (bool) – wait for wait complete.

abstract remove_all_observers()¶: Remove all observers from PyNotiCenter

abstract remove_observer(name: str, fn: Callable, receiver: Optional[Any] = None)¶

Remove observer from PyNotiCenter

Parameters:

name (str) – notification name
fn (Callable) – callback function
receiver (Any) – receiver object

abstract remove_observers(receiver: Any)¶

Remove observers from PyNotiCenter

Parameters:: receiver (Any) – receiver object

abstract shutdown(wait: bool)¶

shutdown PyNotiCenter. Once shutdown, you can’t call it anymore.

When shutdown is called, you can not post new task to task queue, all task queue will ignore the new task. When wait is setted, it will block until all task complete, if not set, all task queue will shutdown as fast as possible.

Parameters:: wait (bool) – set wait for task complete or not.

abstract wait_until_task_complete()¶: wait until all task complete. it will block until there is no task.

PyNotiOptions¶

class PyNotiOptions(queue: str, fn_with_task_id: bool = False)¶: Bases: object

PyNotiTask¶

PyNotiTaskQueue¶

class PyNotiTaskQueue(name: str, scheduler_runloop: AbstractEventLoop, thread_pool: ThreadPoolExecutor)¶

Bases: object

PyNotiTaskQueue, each task queue has its own thread. All function thread safety