prozorro_sale.tools.application module

class prozorro_sale.tools.application.AioHttpWrapper(aiohttp_wrapper_args: dict, loop: asyncio.events.AbstractEventLoop, stop_callback=None, graceful_exit_time=False)

Bases: prozorro_sale.tools.application.CoroutineWrapper

Wrapper for starting the application asynchronously or serving on multiple HOST/PORT. .. rubric:: Example

>>> runner = web.AppRunner(app)
... await runner.setup()
... site = web.TCPSite(runner, 'localhost', 8080)
... await site.start()

__init__(aiohttp_wrapper_args: dict, loop: asyncio.events.AbstractEventLoop, stop_callback=None, graceful_exit_time=False)

Initialize web.AppRunner

Variables

• loop (asyncio.AbstractEventLoop) – current Event loop

• runner (web.AppRunner) – aiohttp web app runner

• _shutdown (bool) – application stop flag

async __sleep_stream()

Creates HTTP server

__slots__ = ('aiohttp_wrapper_args', 'runner')

async _cleanup()

Cleanup web.AppRunner.

aiohttp_wrapper_args

runner

class prozorro_sale.tools.application.ApplicationWrapper(loop: Optional[asyncio.events.AbstractEventLoop] = None, return_when: Optional[str] = 'FIRST_EXCEPTION', signals: Optional[Union[list, set]] = None, debug: Optional[bool] = None)

Bases: object

Application Wrapper allows to create and configure app with preset configuration, and also has custom methods.

Custom methods:

• add_web_app (add custom aiohttp web application)

• add_coroutine (add task to asyncio.loop)

ALL_COMPLETED = 'ALL_COMPLETED'

condition to exit after all coroutines complete

Type

str

FIRST_COMPLETED = 'FIRST_COMPLETED'

condition to exit after the first coroutines complete

Type

str

FIRST_EXCEPTION = 'FIRST_EXCEPTION'

condition to exit after the first error occurs

Type

str

__call__()

Starts configurated ApplicationWrapper

__cancel_tasks() → None

Method for cancel all active tasks.

__close_loop()

Close aiohttp.loop.

__handle_signal(_signal_name, callback)

Set callback as the handler for the _signal_name signal. The callback will be invoked by loop, along with other queued callbacks and runnable coroutines of that event loop.

Returns

None

__init__(loop: Optional[asyncio.events.AbstractEventLoop] = None, return_when: Optional[str] = 'FIRST_EXCEPTION', signals: Optional[Union[list, set]] = None, debug: Optional[bool] = None)

Application wrapper

Parameters

• loop (asyncio.AbstractEventLoop, optional) – Event loop

• return_when (str, optional) – wrapper stop condition, default FIRST_EXCEPTION

• signals (list, optional) – UNIX signals to stop

• debug (bool, optional) – run loop on debug mode

Variables

• runner (asyncio.Task) – current wrapper task

• loop (asyncio.AbstractEventLoop`) – current Event loop

• return_when (str) – exit condition

• apps (list) – list of registered applications

Returns

None

__shutdown_app()

Method for shutdown all added apps.

__shutdown_runner()

Method for shutdown web.AppRunner.

__slots__ = ('loop', 'apps', 'return_when', 'runner', 'signals')

async __wait(tasks, return_when=None, canceled_pending=False)

Wrapper for asyncio.wait.

Parameters

• tasks (list) – coroutine tasks

• return_when – asyncio.FIRST_EXCEPTION, asyncio.FIRST_COMPLETED, asyncio.ALL_COMPLETED

• canceled_pending (bool) – If True cancel pending tasks

_graceful_exit(*args, **kwargs) → None

Method for graceful close aiohttp.loop. if loop is running, wait until complete.

add_coroutine(coro, stop_callback=None, graceful_exit_time=False)

Method for add task to asyncio.loop

Parameters

• coro –

• stop_callback –

• graceful_exit_time –

add_web_app(app: Union[aiohttp.web_app.Application, Awaitable[aiohttp.web_app.Application]], *, host: Optional[Union[str, Iterable[str]]] = '0.0.0.0', port: Optional[int] = 80, shutdown_timeout: Optional[float] = 60.0, keepalive_timeout: Optional[float] = 75.0, ssl_context: Optional[ssl.SSLContext] = None, backlog: Optional[int] = 128, access_log_class: Type[aiohttp.abc.AbstractAccessLogger] = <class 'aiohttp.web_log.AccessLogger'>, reuse_address: Optional[bool] = True, reuse_port: Optional[bool] = True, stop_callback: Optional[bool] = None, graceful_exit_time: Optional[bool] = False) → None

Method for add custom aiohttp web application.

Parameters

• app (web.Application) – aiohttp Application

• host (str, optional) – listen ip address, default ‘0.0.0.0’

• port (int, optional) – listen tcp port, default 80

• shutdown_timeout (float, optional) –

• keepalive_timeout (float, optional) –

• ssl_context (web.SSLContext, optional) –

• backlog (int, optional) –

• access_log_class (web.AbstractAccessLogger, optional) –

• reuse_address (bool, optional) –

• reuse_port (bool, optional) –

• stop_callback (bool, optional) –

• graceful_exit_time (bool, optional) –

add_worker_app(app, stop_callback=None, graceful_exit_time=False)

apps

loop

return_when

runner

signals

class prozorro_sale.tools.application.CoroutineWrapper(coroutine, loop: asyncio.events.AbstractEventLoop, stop_callback=None, graceful_exit_time=False)

Bases: object

Coroutine wrapper.

__init__(coroutine, loop: asyncio.events.AbstractEventLoop, stop_callback=None, graceful_exit_time=False)

Parameters

• coroutine –

• loop (asyncio.AbstractEventLoop`) – current Event loop

• stop_callback –

• graceful_exit_time –

Variables

• task (asyncio.Task) – current application task

• loop (asyncio.AbstractEventLoop`) – current Event loop

__repr__()

Return repr(self).

__slots__ = ('loop', 'task', '_stop_callback', '_shutdown', '_graceful_exit_time')

__str__()

Return str(self).

_cancel()

Cancel asynchronous task.

async _cleanup()

_graceful_exit_time

_shutdown

_stop_callback

initialize()

Create asynchronous task.

loop

shutdown()

Cleanup asynchronous loop.

task

class prozorro_sale.tools.application.WorkerWrapper(coroutine, loop: asyncio.events.AbstractEventLoop, stop_callback=None, graceful_exit_time=False)

Bases: prozorro_sale.tools.application.CoroutineWrapper

_graceful_exit_time

_shutdown

_stop_callback

loop

task