pawnlib.asyncio package

from pawnlib.asyncio import *

Submodules

pawnlib.asyncio.run module

from pawnlib.asyncio import run

class AsyncTasks(max_at_once=10, max_per_second=10, title='Working on async tasks ...', debug=False, status=False, **kwargs)[source]

Bases: object

This Class is to run asyncio using aiometer.

Parameters:

max_at_once (int, optional) – Limit maximum number of concurrently running tasks. |default| 10
max_per_second (int, optional) – Limit request rate to not overload the server. |default| 10
title (str, optional) – Title of the tasks |default| 'Working on async tasks ...'
status (bool, optional) – Status of the tasks |default| False
debug (bool, optional) – Whether to use debug |default| False
kwargs –

Example

async_tasks = AsyncTasks(max_at_once=10, max_per_second=10)
async_tasks.generate_tasks(
    target_list=range(1, 100) ,
    function=run_container,
    **{"args": args}
).run()

__init__(max_at_once=10, max_per_second=10, title='Working on async tasks ...', debug=False, status=False, **kwargs)[source]

This Class is to run asyncio using aiometer.

Parameters:

max_at_once (int, optional) – Limit maximum number of concurrently running tasks. |default| 10
max_per_second (int, optional) – Limit request rate to not overload the server. |default| 10
title (str, optional) – Title of the tasks |default| 'Working on async tasks ...'
status (bool, optional) – Status of the tasks |default| False
debug (bool, optional) – Whether to use debug |default| False
kwargs –

Example

async_tasks = AsyncTasks(max_at_once=10, max_per_second=10)
async_tasks.generate_tasks(
    target_list=range(1, 100) ,
    function=run_container,
    **{"args": args}
).run()

get_tasks()[source]

generate_tasks(target_list=None, function=None, **kwargs)[source]

This function generates the async tasks list

Parameters:

target_list (optional) – List of targets for asynchronous execution |default| None
function (optional) – Name of the function to execute |default| None
kwargs –

Returns:

run()[source]

This function executes an asynchronous operation.

Returns:

async_partial(async_fn, *args, **kwargs)[source]

run_in_async_loop(f)[source]

class AsyncHttp(max_at_once=10, max_per_second=10, title='Working on async tasks ...', debug=False, status=False, urls=None, **kwargs)[source]

Bases: AsyncTasks

This class is a subclass of AsyncTasks and is used to handle asynchronous HTTP requests.

max_at_once

Maximum number of tasks to run at once.

Type:: int

max_per_second

Maximum number of tasks to start per second.

Type:: int

title

Title for the progress bar.

Type:: str

debug

If True, print debug information.

Type:: bool

status

If True, print status information.

Type:: bool

urls

List of URLs to fetch.

Type:: list

This Class is to run asyncio using aiometer.

Parameters:

max_at_once (int, optional) – Limit maximum number of concurrently running tasks. |default| 10
max_per_second (int, optional) – Limit request rate to not overload the server. |default| 10
title (optional) – Title of the tasks |default| 'Working on async tasks ...'
status (bool, optional) – Status of the tasks |default| False
debug (bool, optional) – Whether to use debug |default| False
kwargs –

Example

async_tasks = AsyncTasks(max_at_once=10, max_per_second=10)
async_tasks.generate_tasks(
    target_list=range(1, 100) ,
    function=run_container,
    **{"args": args}
).run()

append_task(task)[source]

Append a new task to the task list.

Parameters:: task (str or dict) – If it’s a string, it’s considered as a URL. If it’s a dictionary, it should contain a ‘url’ key.

async fetch_httpx_url(url, method='get', timeout=4, info='', max_keepalive_connections=10, max_connections=20, **kwargs)[source]

Asynchronously fetch a URL using httpx.

Parameters:

url (str) – The URL to fetch.
method (str) – The HTTP method to use. Default is “get”. |default| 'get'
timeout (int) – The timeout for the request in seconds. Default is 4. |default| 4
info (str) – Additional information for the request. Default is an empty string. |default| ''
max_keepalive_connections (int) – The maximum number of keep-alive connections. Default is 10. |default| 10
max_connections (int) – The maximum number of connections. Default is 20. |default| 20
**kwargs – Additional keyword arguments for the httpx request.

Returns:

The response from the server, or an empty dictionary if an error occurred.

Return type:

httpx.Response or dict

pawnlib.asyncio.async_helper module

from pawnlib.asyncio import async_helper

async shutdown_async_tasks(loop=None, tasks=None, logger=None, exit_on_shutdown=True, verbose=1, exit_code=0)[source]

Shutdown all pending async tasks and close the loop gracefully.

Parameters:

loop (asyncio.AbstractEventLoop) – The event loop to use. If None, the current running loop will be used. |default| None
tasks (List[asyncio.Task]) – List of tasks to cancel. If None, all pending tasks in the loop will be cancelled. |default| None
logger (logging.Logger) – Logger instance for logging. If None, print will be used as fallback. |default| None
exit_on_shutdown (bool) – If True, the system will exit after shutdown. |default| True
verbose (int) – The verbosity level for logging output. |default| 1
exit_code (int) – Exit code to return when exiting the system (default is 0). |default| 0