ppqueue.PPQueue¶

A parallel processing job runner / data structure.

PARAMETER	DESCRIPTION
`max_concurrent`	max number of concurrently running jobs. TYPE: `int` DEFAULT: `cpu_count()`
`max_size`	max size of the queue (default=0, unlimited). TYPE: `int` DEFAULT: `0`
`engine`	the engine used to run jobs. TYPE: `str \| type[Process] \| type[Process] \| type[Thread]` DEFAULT: `Process`
`name`	an identifier for this queue. TYPE: `str \| None` DEFAULT: `None`
`callback`	a callable that is called immediately after each job is finished. TYPE: `Callable[[Job], Any] \| None` DEFAULT: `None`
`show_progress`	global setting for showing progress bars. TYPE: `bool` DEFAULT: `False`
`drop_finished`	if True, the queue will not store finished jobs for retrieval. TYPE: `bool` DEFAULT: `False`
`stop_when_idle`	if True, the queue will stop the pulse when all jobs are finished. TYPE: `bool` DEFAULT: `False`
`pulse_freq_ms`	the interval at which jobs are transitioned between internal queues. TYPE: `int` DEFAULT: `100`
`no_start`	if True, do not start the queue pulse on instantiation. TYPE: `bool` DEFAULT: `False`

Examples:

from ppqueue import PPQueue
from time import sleep

with PPQueue() as queue:
    jobs = queue.map(sleep, [1, 2, 3, 4, 5])

assert len(jobs) == 5

Functions¶

collect ¶

collect(
    n: int = 0, wait: bool = False, **kwargs: Any
) -> list[Job]

Removes and returns all finished jobs from the queue.

PARAMETER	DESCRIPTION
`n`	collect this many jobs (default=0, all) TYPE: `int` DEFAULT: `0`
`wait`	If True, block until this many jobs are finished. Else, immediately return all finished. TYPE: `bool` DEFAULT: `False`
`**kwargs`	kwargs given to `PPQueue.wait`. TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`list[Job]`	a list of `Job` instances.

Examples:

from ppqueue import PPQueue


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    for i in range(5):
        _ = queue.enqueue(add_nums, args=[i, 100])

    jobs = queue.collect(wait=True)

assert isinstance(jobs, list)

assert len(jobs) == 5

dequeue ¶

dequeue(
    *,
    wait: bool = False,
    _peek: bool = False,
    **kwargs: Any
) -> Job | None

Removes and returns the finished job with the highest priority from the queue.

PARAMETER	DESCRIPTION
`wait`	if no jobs are finished, wait for one. TYPE: `bool` DEFAULT: `False`
`**kwargs`	passed to `PPQueue.wait` TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`Job \| None`	...

Examples:

from ppqueue import PPQueue


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    for i in range(5):
        _ = queue.enqueue(add_nums, args=[i, 100])

    jobs = [queue.dequeue(wait=True) for _ in range(queue.size())]

results = [job.result for job in jobs]

assert [job.result for job in jobs] == [100, 101, 102, 103, 104]

dispose ¶

dispose() -> None

Stop running jobs, then clear the queue, then stop the queue pulse.

enqueue ¶

enqueue(
    fun: Callable[..., Any],
    /,
    args: Any = None,
    kwargs: dict[str, Any] | None = None,
    name: str | None = None,
    priority: int = 100,
    group: int | None = None,
    timeout: float = 0,
    suppress_errors: bool = False,
    skip_on_group_error: bool = False,
) -> int

Adds a job to the queue.

PARAMETER	DESCRIPTION
`fun`	... TYPE: `Callable[..., Any]`
`args`	... TYPE: `Any` DEFAULT: `None`
`kwargs`	... TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`name`	... TYPE: `str \| None` DEFAULT: `None`
`priority`	... TYPE: `int` DEFAULT: `100`
`group`	... TYPE: `int \| None` DEFAULT: `None`
`timeout`	... TYPE: `float` DEFAULT: `0`
`suppress_errors`	... TYPE: `bool` DEFAULT: `False`
`skip_on_group_error`	... TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`int`	...

Examples:

from ppqueue import PPQueue


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    for i in range(5):
        _ = queue.enqueue(add_nums, args=[i, 100])

    jobs = queue.collect(wait=True)

results = [job.result for job in jobs]

assert results == [100, 101, 102, 103, 104]

is_busy ¶

is_busy() -> bool

True if max concurrent limit is reached or if there are waiting jobs.

RETURNS	DESCRIPTION
`bool`	...

is_empty ¶

is_empty() -> bool

True if there are no jobs in the queue system.

RETURNS	DESCRIPTION
`bool`	...

is_full ¶

is_full() -> bool

True if the number of jobs in the queue system is equal to max_size.

RETURNS	DESCRIPTION
`bool`	...

map ¶

map(
    fun: Callable[..., Any],
    iterable: Sequence[Any],
    /,
    *args: Any,
    timeout: float = 0,
    show_progress: bool | None = None,
    **kwargs: Any,
) -> list[Job]

Submits many jobs to the queue -- one for each item in the iterable. Waits for all to finish, then returns the results.

PARAMETER	DESCRIPTION
`fun`	... TYPE: `Callable[..., Any]`
`iterable`	... TYPE: `Sequence[Any]`
`*args`	... TYPE: `Any` DEFAULT: `()`
`timeout`	... TYPE: `float` DEFAULT: `0`
`show_progress`	... TYPE: `bool \| None` DEFAULT: `None`
`**kwargs`	... TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`list[Job]`	...

Examples:

from ppqueue import PPQueue
from time import sleep

with PPQueue() as queue:
    jobs = queue.map(sleep, [1, 2, 3, 4, 5])

assert len(jobs) == 5

peek ¶

peek(*args: Any, **kwargs: Any) -> Job | None

Returns the job with the highest priority from the queue.

Similar to dequeue / pop, but the job remains in the queue.

PARAMETER	DESCRIPTION
`*args`	... TYPE: `Any` DEFAULT: `()`
`**kwargs`	... TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`Job \| None`	...

Examples:

from ppqueue import PPQueue


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    for i in range(5):
        _ = queue.put(add_nums, args=[i, 100])

    print("Before:", queue.size())

    job = queue.peek(wait=True)

    print("After:", queue.size())

    # Before: 5
    # After: 5

assert job.result == 100

pop ¶

pop(*args: Any, **kwargs: Any) -> Job | None

Alias for dequeue.

PARAMETER	DESCRIPTION
`*args`	... TYPE: `Any` DEFAULT: `()`
`**kwargs`	... TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`Job \| None`	...

Examples:

from ppqueue import PPQueue


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    for i in range(5):
        _ = queue.put(add_nums, args=[i, 100])

    jobs = [queue.pop(wait=True) for _ in range(queue.size())]

results = [job.result for job in jobs]

assert results == [100, 101, 102, 103, 104]

put ¶

put(*args, **kwargs) -> int

Alias for enqueue.

Adds a job to the queue.

PARAMETER	DESCRIPTION
`*args`	... DEFAULT: `()`
`**kwargs`	... DEFAULT: `{}`

RETURNS	DESCRIPTION
`int`	...

Examples:

from ppqueue import PPQueue


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    for i in range(5):
        _ = queue.put(add_nums, args=[i, 100])

    jobs = queue.collect(wait=True)

results = [job.result for job in jobs]

assert results == [100, 101, 102, 103, 104]

size ¶

size(
    *,
    waiting: bool = False,
    working: bool = False,
    finished: bool = False
) -> int

Get the number of jobs in the queue in state: waiting, working, and/or finished.

If no options are given, the total number of jobs in the queue is returned.

PARAMETER	DESCRIPTION
`waiting`	include waiting jobs. TYPE: `bool` DEFAULT: `False`
`working`	include working jobs. TYPE: `bool` DEFAULT: `False`
`finished`	include finished jobs. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`int`	...

Examples:

from ppqueue import PPQueue


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    for i in range(5):
        _ = queue.enqueue(add_nums, args=[i, 100])
        print(queue.size())
# 1
# 2
# 3
# 4
# 5

starmap ¶

starmap(
    fun: Callable[..., Any],
    iterable: Sequence[Sequence[Any]],
    /,
    *args: Any,
    timeout: float = 0,
    show_progress: bool | None = None,
    **kwargs: Any,
) -> list[Job]

Submits many jobs to the queue -- one for each sequence in the iterable. Waits for all to finish, then returns the results.

PARAMETER	DESCRIPTION
`fun`	... TYPE: `Callable[..., Any]`
`iterable`	... TYPE: `Sequence[Sequence[Any]]`
`*args`	static arguments passed to the function. TYPE: `Any` DEFAULT: `()`
`timeout`	... TYPE: `float` DEFAULT: `0`
`show_progress`	... TYPE: `bool \| None` DEFAULT: `None`
`**kwargs`	static keyword-arguments passed to the function. TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`list[Job]`	...

Examples:

from ppqueue import PPQueue


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    jobs = queue.starmap(add_nums, [(1, 2), (3, 4)])

results = [job.result for job in jobs]

assert results == [3, 7]

starmapkw ¶

starmapkw(
    fun: Callable[..., Any],
    iterable: Sequence[dict[str, Any]],
    /,
    *args: Any,
    timeout: float = 0,
    show_progress: bool | None = None,
    **kwargs: Any,
) -> list[Job]

Submits many jobs to the queue -- one for each dictionary in the iterable. Waits for all to finish, then returns the results.

PARAMETER	DESCRIPTION
`fun`	... TYPE: `Callable[..., Any]`
`iterable`	... TYPE: `Sequence[dict[str, Any]]`
`*args`	static arguments passed to the function. TYPE: `Any` DEFAULT: `()`
`timeout`	... TYPE: `float` DEFAULT: `0`
`show_progress`	... TYPE: `bool \| None` DEFAULT: `None`
`**kwargs`	static keyword-arguments passed to the function. TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`list[Job]`	...

Examples:

from ppqueue import PPQueue
from time import sleep


def add_nums(x: int, y: int) -> int:
    return x + y


with PPQueue() as queue:
    jobs = queue.starmapkw(add_nums, [{"x": 1, "y": 2}, {"x": 3, "y": 4}])

results = [job.result for job in jobs]

assert results == [3, 7]

start ¶

start() -> None

Start the queue pulse.

stop ¶

stop() -> None

Stop the queue pulse.

wait ¶

wait(
    *,
    n: int = 0,
    timeout: float = 0,
    poll_ms: int = 0,
    show_progress: bool | None = None
) -> int

Wait for jobs to finish.

PARAMETER	DESCRIPTION
`n`	the number of jobs to wait for (default=0, all). TYPE: `int` DEFAULT: `0`
`timeout`	seconds to wait before raising `TimeoutError` (default=0, indefinitely). TYPE: `float` DEFAULT: `0`
`poll_ms`	milliseconds to pause between checks (default=100). TYPE: `int` DEFAULT: `0`
`show_progress`	if True, present a progress bar. TYPE: `bool \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`int`	If `n <= 0`, returns the count of unfinished jobs.
`int`	Else, returns the count of finished jobs.