aiomisc - miscellaneous utils for asyncio
=========================================

.. image:: https://coveralls.io/repos/github/aiokitchen/aiomisc/badge.svg?branch=master
   :target: https://coveralls.io/github/aiokitchen/aiomisc
   :alt: Coveralls

.. image:: https://github.com/aiokitchen/aiomisc/actions/workflows/tests.yml/badge.svg
   :target: https://github.com/aiokitchen/aiomisc/actions/workflows/tests.yml
   :alt: Actions

.. image:: https://img.shields.io/pypi/v/aiomisc.svg
   :target: https://pypi.python.org/pypi/aiomisc/
   :alt: Latest Version

.. image:: https://img.shields.io/pypi/wheel/aiomisc.svg
   :target: https://pypi.python.org/pypi/aiomisc/

.. image:: https://img.shields.io/pypi/pyversions/aiomisc.svg
   :target: https://pypi.python.org/pypi/aiomisc/

.. image:: https://img.shields.io/pypi/l/aiomisc.svg
   :target: https://pypi.python.org/pypi/aiomisc/


As a programmer, you are no stranger to the challenges that come with building
and maintaining software applications. One area that can be particularly
difficult is designing the architecture of asynchronous I/O software.

This is where ``aiomisc`` comes in. It is a Python library that
provides a collection of utility functions and classes for working with
asynchronous I/O in a more intuitive and efficient way. It is built on
top of the ``asyncio`` library and is designed to make it easier for
developers to write asynchronous code that is both reliable and scalable.

With ``aiomisc``, you can take advantage of powerful features like
:doc:`worker pools </worker_pool>`, :doc:`connection pools </pool>`,
:doc:`circuit breaker pattern </circuit_breaker>`,
and retry mechanisms such as :doc:`asyncbackoff </async_backoff>` and
:ref:`asyncretry <asyncretry>` to make your asyncio code more robust and
easier to maintain. In this documentation, we'll take a closer look at what
``aiomisc`` has to offer and how it can help you streamline your asyncio
service development.

Why use aiomisc?
----------------

**Problem:** Production asyncio applications require significant boilerplate
for logging, graceful shutdown, thread pools, and error handling.

**Solution:** aiomisc handles infrastructure so you can focus on business logic.

+---------------------------+---------------------------+
| Plain asyncio             | With aiomisc              |
+===========================+===========================+
| Manual signal handling    | Built into entrypoint     |
+---------------------------+---------------------------+
| Manual logging setup      | Single parameter          |
+---------------------------+---------------------------+
| Manual thread pool        | Automatic + @threaded     |
+---------------------------+---------------------------+
| Try/finally cleanup       | Service stop() method     |
+---------------------------+---------------------------+

Installation
------------

Installation is possible in standard ways, such as PyPI or installation from
a git repository directly.

Installing from PyPI_:

.. code-block:: bash

    pip3 install aiomisc

Installing from github.com:

.. code-block:: bash

    # Using git tool
    pip3 install git+https://github.com/aiokitchen/aiomisc.git

    # Alternative way using http
    pip3 install \
        https://github.com/aiokitchen/aiomisc/archive/refs/heads/master.zip

The package contains several extras and you can install additional dependencies
if you specify them in this way.

With uvloop_:

.. code-block:: bash

    pip3 install "aiomisc[uvloop]"


With aiohttp_:

.. code-block:: bash

    pip3 install "aiomisc[aiohttp]"

Complete table of extras below:

+-----------------------------------+------------------------------------------------+
| example                           |  description                                   |
+===================================+================================================+
| ``pip install aiomisc[aiohttp]``  | For running aiohttp_ applications.             |
+-----------------------------------+------------------------------------------------+
| ``pip install aiomisc[asgi]``     | For running ASGI_ applications                 |
+-----------------------------------+------------------------------------------------+
| ``pip install aiomisc[carbon]``   | Sending metrics to carbon_ (part of graphite_) |
+-----------------------------------+------------------------------------------------+
| ``pip install aiomisc[cron]``     | use croniter_ for scheduling tasks             |
+-----------------------------------+------------------------------------------------+
| ``pip install aiomisc[raven]``    | Sending exceptions to sentry_ using raven_     |
+-----------------------------------+------------------------------------------------+
| ``pip install aiomisc[rich]``     | Use rich_ for logging                          |
+-----------------------------------+------------------------------------------------+
| ``pip install aiomisc[uvicorn]``  | For running ASGI_ application using uvicorn_   |
+-----------------------------------+------------------------------------------------+
| ``pip install aiomisc[uvloop]``   | use uvloop_ as a default event loop            |
+-----------------------------------+------------------------------------------------+

.. _ASGI: https://asgi.readthedocs.io/en/latest/
.. _PyPI: https://pypi.org/
.. _aiohttp: https://pypi.org/project/aiohttp
.. _carbon: https://pypi.org/project/carbon
.. _croniter: https://pypi.org/project/croniter
.. _graphite: http://graphiteapp.org
.. _raven: https://pypi.org/project/raven
.. _rich: https://pypi.org/project/rich
.. _sentry: https://sentry.io/
.. _uvloop: https://pypi.org/project/uvloop
.. _uvicorn: https://pypi.org/project/uvicorn

You can combine extras values by separating them with commas, for example:

.. code-block:: bash

    pip3 install "aiomisc[aiohttp,cron,rich,uvloop]"


Quick Start
-----------

This section will cover how this library creates and uses the event loop and
creates services. For more details see :doc:`/tutorial` section, and you can
always refer to the :doc:`/modules` and :doc:`/api/index` sections for help.

Event-loop and entrypoint
+++++++++++++++++++++++++

Let's look at this simple example first:

.. code-block:: python

    import asyncio
    import logging

    import aiomisc

    log = logging.getLogger(__name__)

    async def main():
        log.info('Starting')
        await asyncio.sleep(3)
        log.info('Exiting')


    if __name__ == '__main__':
        with aiomisc.entrypoint(log_level="info", log_format="color") as loop:
            loop.run_until_complete(main())


This code declares an asynchronous ``main()`` function that exits after
3 seconds. It would seem nothing interesting, but the whole point is in
the ``entrypoint``.

At first glance the ``entrypoint`` does not do much, it just creates an
event-loop and transfers control to the user. However, under the hood, the
logger is configured in a separate thread, a pool of threads is created,
services are started, but more on that later as there are no services
in this example.

Alternatively, you can use the standard ``asyncio.Runner``:

.. code-block:: python
    :name: test_index_runner

    import asyncio

    async def main():
        await asyncio.sleep(1)

    if __name__ == '__main__':
        with asyncio.Runner() as runner:
            runner.run(main())

Services
++++++++

The main thing that an ``entrypoint`` does is start and gracefully
stop services.

The service concept within this library means a class derived from
the ``aiomisc.Service`` class and implementing the
``async def start(self) -> None:`` method and optionally the
``async def stop(self, exc: Optional[ Exception]) -> None`` method.

The concept of stopping a service doesn't necessarily mean pressing ``Ctrl+C``
by the user, it's actually just exiting the ``entrypoint`` context manager.

The example below shows what your service might look like:

.. code-block:: python

    from aiomisc import entrypoint, Service

    class MyService(Service):
        async def start(self):
            do_something_when_start()

        async def stop(self, exc):
            do_graceful_shutdown()


    with entrypoint(MyService()) as loop:
        loop.run_forever()

The entry point can start as many instances of the service as it likes,
and all of them will start concurrently.

There is also a way if the ``start`` method is a payload for a service,
and then there is no need to implement the stop method, since the running
task with the ``start`` function will be canceled at the stop stage.
But in this case, you will have to notify the ``entrypoint`` that the
initialization of the service instance is complete and it can continue.

Like this:

.. code-block:: python

    import asyncio
    from threading import Event
    from aiomisc import entrypoint, Service

    event = Event()

    class MyService(Service):
        async def start(self):
            # Send signal to entrypoint for continue running
            self.start_event.set()
            await asyncio.sleep(3600)


    with entrypoint(MyService()) as loop:
        assert event.is_set()

.. note::

    The ``entrypoint`` passes control to the body of the context manager only
    after all service instances have started. As mentioned above, a start is
    considered to be the completion of the ``start`` method or the setting of
    an start event with ``self.start_event.set()``.

The whole power of this library is in the set of already implemented or
abstract services.
Such as: :ref:`AIOHTTPService <aiohttp-service>`,
:ref:`ASGIService <asgi-service>`, :ref:`TCPServer <tcp-server>`,
:ref:`UDPServer <udp-server>`, :ref:`TCPClient <tcp-client>`,
:ref:`PeriodicService <periodic-service>`,
:ref:`CronService <cron-service>` and so on.

Unfortunately in this section it is not possible to pay more attention to this,
please pay attention to the :doc:`/tutorial` section section, there are more
examples and explanations, and of course you can always find an answer in
the :doc:`/api/index` or in the source code. The authors have tried to make
the source code as clear and simple as possible, so feel free to explore it.


Versioning
----------

This software follows `Semantic Versioning`_


Summary: it's given a version number MAJOR.MINOR.PATCH, increment the:

* MAJOR version when you make incompatible API changes
* MINOR version when you add functionality in a backwards compatible manner
* PATCH version when you make backwards compatible bug fixes
* Additional labels for pre-release and build metadata are available as
  extensions to the MAJOR.MINOR.PATCH format.

In this case, the package version is assigned automatically with poem-plugins_,
it using on the tag in the repository as a major and minor and the counter,
which takes the number of commits between tag to the head of branch.

.. _poem-plugins: https://pypi.org/project/poem-plugins


How to develop?
---------------

This project, like most open source projects, is developed by enthusiasts,
you can join the development, submit issues, or send your merge requests.

In order to start developing in this repository, you need to do the
following things.

Should be installed:

* Python 3.7+ as ``python3``
* Installed Poetry_ as ``poetry``

.. _Poetry: https://python-poetry.org/docs/

For setting up developer environment just execute:

    .. code-block::

        # installing all dependencies
        poetry install

        # setting up pre-commit hooks
        poetry run pre-commit install

        # adding poem-plugins to the poetry
        poetry self add poem-plugins


.. _Semantic Versioning: http://semver.org/


.. toctree::
   :glob:
   :numbered:
   :maxdepth: 3

   tutorial
   modules
   api/index


---

Tutorial
========


``aiomisc`` is a Python library that provides a set of utilities for building
asynchronous services. It allows you to split your program into smaller,
independent services that can run concurrently, improving the overall
performance and scalability of your application.

The main approach in this library is to split your program into independent
services that can work concurrently in asynchronous mode. The library also
provides a set of ready-to-use services with pre-written start and stop logic.

The vast majority of functions and classes are written in such a way that
they can be used in a program that was not originally designed according
to the principles outlined in this manual. This means that if you don't
plan to modify your code too much, but only use a few useful functions or
classes, then everything should work.

Overall, aiomisc is a powerful tool for developers looking to build efficient
and scalable asynchronous services in Python.

Services
++++++++

If you want to run a tcp or web server, you will
have to write something like this:

.. code-block:: python

    import asyncio

    async def example_async_func():
        # do my async business logic
        await init_db()
        await init_cache()
        await start_http_server()
        await start_metrics_server()

    loop = asyncio.get_event_loop()
    loop.run_until_complete(example_async_func())
    # Continue running all background tasks
    loop.run_forever()

In order to start or stop an async programs, usually using function
``asyncio.run(example_async_func())`` which available since Python 3.7.
This function takes an instance of a coroutine and cancels
all still running tasks before returning a result.
To continue executing the code indefinitely,
you can perform the following trick:

.. code-block:: python

    import asyncio

    async def example_async_func():
        # do my async business logic
        await init_db()
        await init_cache()
        await start_http_server()
        await start_metrics_server()

        # Make future which never be done
        # using instead loop.run_forever()
        await asyncio.Future()

    asyncio.run(example_async_func())


When the user presses `Ctrl+C`, the program simply terminates, but if you
want to explicitly free up some resources, for example, close database
connections or rolling back incomplete transactions, then you have to
do something like this:

.. code-block:: python

    import asyncio

    async def example_async_func():
        try:
            # do my async business logic
            await init_db()
            await init_cache()
            await start_http_server()
            await start_metrics_server()

            # Make future which never be done
            # using instead loop.run_forever()
            await asyncio.Future()
        except asyncio.CancelledError:
            # Do this block when SIGTERM has been received
            pass
        finally:
            # Do this block on exit
            ...

    asyncio.run(example_async_func())


It is a good solution because it is implemented without any 3rd-party libraries.
When your program starts to grow, you will probably want to optimize the
startup time in a simple way, namely to do all initialization concurrently.
At first glance it seems that this code will solve the problem:


.. code-block:: python

    import asyncio

    async def example_async_func():
        try:
            # do my async business logic
            await asyncio.gather(
                init_db(),
                init_cache(),
                start_http_server(),
                start_metrics_server(),
            )

            # Make future which never be done
            # using instead loop.run_forever()
            await asyncio.Future()
        except asyncio.CancelledError:
            # Do this block when SIGTERM has been received
            pass
        finally:
            # Do this block on exit
            ...

    asyncio.run(example_async_func())

But if suddenly some part of the initialization does not go according to plan,
then you somehow have to figure out what exactly went wrong, so with concurrent
execution, the code will no longer be as simple as in this example.

And in order to somehow organize the code, you should make
a separate function that will contain the ``try/except/finally`` block and
contain error handling.


.. code-block:: python

    import asyncio

    async def init_db():
        try:
            # initialize connection
        finally:
            # close connection
            ...

    async def example_async_func():
        try:
            # do my async business logic
            await asyncio.gather(
                init_db(),
                init_cache(),
                start_http_server(),
                start_metrics_server(),
            )

            # Make future which never be done
            # using instead loop.run_forever()
            await asyncio.Future()
        except asyncio.CancelledError:
            # Do this block when SIGTERM has been received
            # TODO: shutdown all things correctly
            pass
        finally:
            # Do this block on exit
            ...

    asyncio.run(example_async_func())


And now if the user presses Ctrl+C, you need to describe the shutdown
logic again, but now in the ``except`` block.

In order to describe the logic of starting and stopping in one place, as well
as testing in one single way, there is a ``Service`` abstraction.

The service is an abstract base class with mandatory ``start()`` and
optional ``stop()`` methods.

The service can operate in two modes. The first is when the ``start()`` method
runs forever, then you do not need to implement a ``stop()``, but you need
to report that the initialization is successfully completed by
setting ``self.start_event.set()``.


.. code-block:: python

    import asyncio
    import aiomisc


    class InfinityService(aiomisc.Service):
        async def start(self):
            # Service is ready
            self.start_event.set()

            while True:
                # do some stuff
                await asyncio.sleep(1)

In this case, stopping the service will consist in the completion of the
coroutine that was created by ``start()``.

The second method is an explicit description of the way
to ``start()`` and ``stop()``.


.. code-block:: python

    import asyncio
    import aiomisc
    from typing import Any


    class OrdinaryService(aiomisc.Service):
        async def start(self):
            # do some stuff
            ...

        async def stop(self, exception: Exception = None) -> Any:
            # do some stuff
            ...

In this case, the service will be started and stopped once.


Service configuration
+++++++++++++++++++++

The ``Service`` is a metaclass, it handles the special attributes of classes
inherited from it at on the their declaration stage.

Here is a simple imperative example of how service initialization can be
extended through inheritance.

.. code-block:: python

    from typing import Any
    import aiomisc

    class HelloService(aiomisc.Service):
        def __init__(self, name: str = "world", **kwargs: Any):
            super().__init__(**kwargs)
            self.name = name

        async def start(self) -> Any:
            print(f"Hello {self.name}")

    with aiomisc.entrypoint(
        HelloService(),
        HelloService(name="Guido")
    ) as loop:
        pass

    # python hello.py
    # <<< Hello world
    # <<< Hello Guido

In fact, you can do nothing of this, since the Service metaclass sets all
the passed keyword parameters to self by default.

.. code-block:: python

    import aiomisc

    class HelloService(aiomisc.Service):
        name: str = "world"

        async def start(self):
            print(f"Hello {self.name}")

    with aiomisc.entrypoint(
        HelloService(),
        HelloService(name="Guido")
    ) as loop:
        pass

    # python hello.py
    # <<< Hello world
    # <<< Hello Guido

If a special class property ``__required__`` is declared, then the service
will required for the user to declare these named parameters.

.. code-block:: python

    import aiomisc

    class HelloService(aiomisc.Service):
        __required__ = ("name", "title")

        name: str
        title: str

        async def start(self):
            await asyncio.sleep(0.1)
            print(f"Hello {self.title} {self.name}")

    with aiomisc.entrypoint(
        HelloService(name="Guido", title="mr.")
    ) as loop:
        pass

Also a very useful special class attribute is ``__async_required__``. It is
useful for writing base classes, in general. This contains the tuple of method
names that must be declared asynchronous explicitly (via ``async def``).

.. code-block:: python

    import aiomisc

    class HelloService(aiomisc.Service):
        __required__ = ("name", "title")
        __async_required__ = ("greeting",)

        name: str
        title: str

        async def greeting(self) -> str:
            await asyncio.sleep(0.1)
            return f"Hello {self.title} {self.name}"

        async def start(self):
            print(await self.greeting())

    class HelloEmojiService(HelloService):
        async def greeting(self) -> str:
            await asyncio.sleep(0.1)
            return f"🙋 {self.title} {self.name}"

    with aiomisc.entrypoint(
        HelloService(name="Guido", title="mr."),
        HelloEmojiService(name="👨", title="🎩")
    ) as loop:
        pass

    # Hello mr. Guido
    # 🙋 🎩 👨

If the inheritor declares these methods differently, there will be an error
at the class declaration stage.

.. code-block:: python

    class BadHello(HelloService):
        def greeting(self) -> str:
            return f"{self.title} {self.name}"

    #Traceback (most recent call last):
    #...
    #TypeError: ('Following methods must be coroutine functions', ('BadHello.greeting',))


dependency injection
++++++++++++++++++++

.. _aiomisc-dependency: https://pypi.org/project/aiomisc-dependency

In some cases, you need to execute some asynchronous code before the service
starts, for example, to pass a database connection to the service instance.
Or if you want to use one instance of some entity for several services.

For such complex configurations, there is `aiomisc-dependency`_ plugin which
is distributed as a independent separate package.

Look at the examples in the documentation, `aiomisc-dependency`_ are
transparently integrates with the ``entrypoint``.

``entrypoint``
++++++++++++++

So the service abstraction is declared, what's next? ``asyncio.run`` does
not know how to work with them, calling them manually has not become easier,
what can this library offer here?

Probably the most magical, complex, and at the same time quite well-tested
code in the library is ``entrypoint``. Initially, the idea of
``entrypoint`` was to get rid of the routine: setting up logs,
setting up a thread pool, as well as starting and stopping services correctly.

Lets check an example:

.. code-block:: python

    import asyncio
    import aiomisc

    ...

    with aiomisc.entrypoint(
        OrdinaryService(),
        InfinityService()
    ) as loop:
        loop.run_forever()

In this example, we will launch the two services described above and continue
execution until the user interrupts them. Next, thanks to the context
manager, we correctly terminate all instances of services.

.. note::

    Entrypoint calls all the ``start()`` methods in all services concurrently,
    and if at least one of them fails, then all services will be stopped.

As mentioned above I just wanted to remove a lot of routine, let's look at the
same example, just pass all the default parameters to the ``entrypoint``
explicitly.

.. code-block:: python

    import asyncio
    import aiomisc


    ...

    with aiomisc.entrypoint(
        OrdinaryService(),
        InfinityService(),
        pool_size=4,
        log_level="info",
        log_format="color",
        log_buffering=True,
        log_buffer_size=1024,
        log_flush_interval=0.2,
        log_config=True,
        policy=asyncio.DefaultEventLoopPolicy(),
        debug=False
    ) as loop:
        loop.run_forever()

Let's not describe what each parameter does. But in general,
``entrypoint`` has create an event-loop, a four threads pool, set
it for the current event-loop, has configure a colored logger with
buffered output, and launched two services.

You can also run the ``entrypoint`` without services,
just configure logging and so on.:

.. code-block:: python

    import asyncio
    import logging
    import aiomisc


    async def sleep_and_exit():
        logging.info("Started")
        await asyncio.sleep(1)
        logging.info("Exiting")


    with aiomisc.entrypoint(log_level="info") as loop:
        loop.run_until_complete(sleep_and_exit())

It is also worth paying attention to the ``aiomisc.run``,
which is similar by its purpose to ``asyncio.run`` while supporting the
start and stop of services and so on.

.. code-block:: python

    import asyncio
    import logging
    import aiomisc


    async def sleep_and_exit():
        logging.info("Started")
        await asyncio.sleep(1)
        logging.info("Exiting")


    aiomisc.run(
        # the first argument
        # is a main coroutine
        sleep_and_exit(),
        # Other positional arguments
        # is service instances
        OrdinaryService(),
        InfinityService(),
        # keyword arguments will
        # be passed as well to the entrypoint
        log_level="info"
    )

.. note::

    As I mentioned above, the library contains lots of already realized
    abstract services that you can use in your project by simply implement
    several methods.

    A full list of services and usage examples can be found on the
    on the :doc:`Services page </services/index>`.

Executing code in thread or process-pools
+++++++++++++++++++++++++++++++++++++++++

.. _working with threads: https://docs.python.org/3/library/asyncio-eventloop.html#executing-code-in-thread-or-process-pools

As explained in `working with threads`_ section in official python
documentation asyncio event loop starts thread pool.

This pool is needed in order to run, for example, name resolution and not
blocks the event loop while low-level ``gethostbyname`` call works.

The size of this thread pool should be configured at application startup,
otherwise you may run into all sorts of problems when this pool is
too large or too small.

By default, the ``entrypoint`` creates a thread pool with size equal to
the number of CPU cores, but not less than 4 and no more than 32 threads.
Of course you can specify as you need.

``@aiomisc.threaded`` decorator
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

The following recommendations for calling blocking functions in threads given
in `working with threads`_ section in official Python documentation:

.. code-block:: python

    import asyncio

    def blocking_io():
        # File operations (such as logging) can block the event loop.
        with open('/dev/urandom', 'rb') as f:
            return f.read(100)

    async def main():
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(None, blocking_io)

    asyncio.run(main())

This library provides a very simple way to do the same:

.. code-block:: python

    import aiomisc

    @aiomisc.threaded
    def blocking_io():
        with open('/dev/urandom', 'rb') as f:
            return f.read(100)

    async def main():
        result = await blocking_io()

    aiomisc.run(main())

As you can see in this example, it is enough to wrap the function
with a decorator ``aiomisc.threaded``, after that it will return an
awaitable object, but the code inside the function will be
sent to the default thread pool.

``@aiomisc.threaded_separate`` decorator
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

If the blocking function runs for a long time, or even indefinitely,
in other words, if the cost of creating a thread is insignificant compared
to the workload, then you can use the decorator ``aiomisc.threaded_separate``.

The decorator starts a new thread not associated with any pool.
Тhe thread will be terminated after the function execution is done.

.. code-block:: python

    import hashlib
    import aiomisc

    @aiomisc.threaded_separate
    def another_one_useless_coin_miner():
        with open('/dev/urandom', 'rb') as f:
            hasher = hashlib.sha256()
            while True:
                hasher.update(f.read(1024))
                if hasher.hexdigest().startswith("0000"):
                    return hasher.hexdigest()

    async def main():
        print(
            "the hash is",
            await another_one_useless_coin_miner()
        )

    aiomisc.run(main())

.. note::

    This approach allows you not to occupy threads in the pool for a long
    time, but at the same time does not limit the number of created threads
    in any way.

More examples you can be found in :doc:`/threads`.

``@aiomisc.threaded_iterable`` decorator
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

If a generator needs to be executed in a thread, there are problems with
synchronization of the thread and the eventloop. This library provides a
custom decorator designed to turn a synchronous generator into an
asynchronous one.

This is very useful if, for example, a queue or database driver has written
synchronous, but you want to use it efficiently in asynchronous code.

.. code-block:: python

    import aiomisc

    @aiomisc.threaded_iterable(max_size=8)
    def urandom_reader():
        with open('/dev/urandom', "rb") as fp:
            while True:
                yield fp.read(8)

    async def main():
        counter = 0
        async for chunk in urandom_reader():
            print(chunk)
            counter += 1
            if counter > 16:
                break

    aiomisc.run(main())

Under the hood, this decorator returns a special object that has a queue, and
asynchronous iterator interface provides access to that queue.

You should always specify the ``max_size`` parameter, which limits the
size of this queue and prevents threaded code from sending too much items to
asynchronous code, in case the asynchronous iteration in case the asynchronous
iteration slacking.

Conclusion
~~~~~~~~~~

On this we need to finish this tutorial, I hope everything was clear here,
and you learned a lot of useful things for yourself. A full description of
the remaining services is presented in the :doc:`/modules` section, or in the
source code. The authors have tried to make the source code as clear and
simple as possible, so feel free to explore it.


---

Modules
+++++++

.. toctree::
   :glob:

   entrypoint
   services/index
   pool
   context
   timeout
   async_backoff
   circuit_breaker
   aggregate
   io
   threads
   process_pool
   utils
   worker_pool
   logging
   pytest
   signal
   plugins
   statistic



---

Working with threads
====================

Why use threads with asyncio?
-----------------------------

Asyncio is designed for non-blocking I/O operations, but many libraries and
operations are inherently blocking:

* File I/O (reading/writing files)
* CPU-intensive computations
* Legacy libraries without async support
* Database drivers without async support
* System calls that don't have async alternatives

Running blocking code directly in an async function blocks the entire event
loop, preventing other coroutines from executing. The solution is to run
blocking code in separate threads while the event loop continues processing
other tasks.

aiomisc provides convenient decorators and utilities to seamlessly integrate
blocking code with asyncio, powered by the `aiothreads`_ library.

.. _aiothreads: https://pypi.org/project/aiothreads/

Quick reference
---------------

+--------------------------------+------------------------------------------------+
| Decorator/Class                | Use case                                       |
+================================+================================================+
| ``@threaded``                  | Run blocking function in thread pool           |
+--------------------------------+------------------------------------------------+
| ``@threaded_separate``         | Run blocking function in new thread            |
+--------------------------------+------------------------------------------------+
| ``@threaded_iterable``         | Async iterate over blocking generator          |
+--------------------------------+------------------------------------------------+
| ``@threaded_iterable_separate``| Same, but on separate thread                   |
+--------------------------------+------------------------------------------------+
| ``IteratorWrapper``            | Wrap existing generator for async iteration    |
+--------------------------------+------------------------------------------------+
| ``sync_await``                 | Call async function from thread                |
+--------------------------------+------------------------------------------------+
| ``FromThreadChannel``          | Send data from thread to async code            |
+--------------------------------+------------------------------------------------+


``@aiomisc.threaded``
---------------------

Wraps a blocking function to run it in the current thread pool. The decorated
function becomes awaitable.

Basic usage
+++++++++++

.. code-block:: python

    import asyncio
    import time
    from aiomisc import threaded


    @threaded
    def blocking_function():
        time.sleep(1)
        return "done"


    async def main():
        # Run two blocking calls in parallel
        results = await asyncio.gather(
            blocking_function(),
            blocking_function(),
        )
        print(results)  # ['done', 'done']


    asyncio.run(main())


Calling modes
+++++++++++++

The ``@threaded`` decorator returns a ``Threaded`` object with multiple
calling methods:

.. code-block:: python

    import time
    from aiomisc import threaded


    @threaded
    def blocking_function():
        time.sleep(0.1)
        return "result"


    async def main():
        # Async call (returns coroutine)
        result = await blocking_function()

        # Explicit async call
        result = await blocking_function.async_call()

        # Synchronous call (blocks current thread)
        result = blocking_function.sync_call()


Works with methods
++++++++++++++++++

The decorator works correctly with instance methods, class methods,
and static methods:

.. code-block:: python

    from aiomisc import threaded


    class MyClass:
        def __init__(self, value):
            self.value = value

        @threaded
        def instance_method(self):
            return self.value

        @threaded
        @classmethod
        def class_method(cls):
            return cls.__name__

        @threaded
        @staticmethod
        def static_method(x):
            return x * 2


    async def main():
        obj = MyClass(42)
        print(await obj.instance_method())       # 42
        print(await MyClass.class_method())      # MyClass
        print(await MyClass.static_method(21))   # 42


``@aiomisc.threaded_separate``
------------------------------

Wraps a blocking function to run it in a new separate thread (not from pool).
Use this for long-running background tasks that would otherwise occupy a
thread pool slot for extended periods.

.. code-block:: python

    import asyncio
    import time
    import threading
    import aiomisc


    @aiomisc.threaded
    def quick_task():
        """Short task - uses thread pool"""
        time.sleep(0.1)
        return "quick"


    @aiomisc.threaded_separate
    def long_running_task(stop_event: threading.Event):
        """Long task - runs in dedicated thread"""
        while not stop_event.is_set():
            print("Working...")
            time.sleep(1)
        return "finished"


    async def main():
        stop_event = threading.Event()

        # Schedule stop after 5 seconds
        loop = asyncio.get_event_loop()
        loop.call_later(5, stop_event.set)

        # Both run concurrently
        results = await asyncio.gather(
            quick_task(),
            long_running_task(stop_event),
        )
        print(results)  # ['quick', 'finished']


    with aiomisc.entrypoint() as loop:
        loop.run_until_complete(main())


Threaded iterators
------------------

``@aiomisc.threaded_iterable``
++++++++++++++++++++++++++++++

Wraps a blocking generator function to make it async-iterable. The generator
runs in a thread pool while yielding values to async code.

.. note::

    The generator uses lazy start - execution begins only when iteration
    starts (first ``async for`` or ``__anext__()`` call), not when entering
    the async context manager.

.. code-block:: python

    import asyncio
    import aiomisc


    @aiomisc.threaded_iterable
    def read_large_file(path):
        """Read file line by line without blocking event loop"""
        with open(path, 'r') as f:
            for line in f:
                yield line.strip()


    async def main():
        async with read_large_file('/etc/hosts') as lines:
            async for line in lines:
                print(line)


    asyncio.run(main())


Buffer size control
^^^^^^^^^^^^^^^^^^^

Use ``max_size`` parameter to control backpressure. The generator thread
will block when the buffer is full:

.. code-block:: python

    import aiomisc


    # Buffer holds at most 2 items
    @aiomisc.threaded_iterable(max_size=2)
    def produce_data():
        for i in range(1000):
            yield i  # Blocks when 2 items buffered


``@aiomisc.threaded_iterable_separate``
+++++++++++++++++++++++++++++++++++++++

Same as ``@threaded_iterable`` but runs in a dedicated thread instead of
the thread pool. Use for long-running generators:

.. code-block:: python

    import aiomisc


    @aiomisc.threaded_iterable_separate
    def tail_file(path):
        """Continuously read new lines from file"""
        with open(path, 'r') as f:
            f.seek(0, 2)  # Go to end
            while True:
                line = f.readline()
                if line:
                    yield line.strip()


Cleanup with context managers
+++++++++++++++++++++++++++++

For infinite generators, always use async context manager or call ``.close()``:

.. code-block:: python

    import aiomisc


    @aiomisc.threaded_iterable(max_size=2)
    def infinite_generator():
        counter = 0
        while True:
            yield counter
            counter += 1


    async def main():
        # Option 1: Context manager (recommended)
        async with infinite_generator() as gen:
            async for value in gen:
                if value >= 10:
                    break  # Context manager handles cleanup

        # Option 2: Manual cleanup
        gen = infinite_generator()
        async for value in gen:
            if value >= 10:
                break
        await gen.close()  # Must call close!


``aiomisc.IteratorWrapper``
---------------------------

Wraps an existing generator to make it async-iterable. Useful when you can't
use decorators or need to specify a custom executor:

.. code-block:: python

    import concurrent.futures
    import aiomisc


    def my_generator():
        for i in range(100):
            yield i


    async def main():
        # Use default thread pool
        wrapper = aiomisc.IteratorWrapper(
            my_generator,
            max_size=10
        )

        async with wrapper as gen:
            async for item in gen:
                print(item)

        # Or use custom thread pool
        pool = concurrent.futures.ThreadPoolExecutor(2)
        wrapper = aiomisc.IteratorWrapper(
            my_generator,
            executor=pool,
            max_size=10
        )

        async with wrapper as gen:
            async for item in gen:
                print(item)

        pool.shutdown()


``aiomisc.IteratorWrapperSeparate``
-----------------------------------

Same as ``IteratorWrapper`` but runs the generator in a dedicated thread:

.. code-block:: python

    import aiomisc


    def blocking_generator():
        while True:
            yield "data"


    async def main():
        wrapper = aiomisc.IteratorWrapperSeparate(
            blocking_generator,
            max_size=5
        )

        async with wrapper as gen:
            count = 0
            async for item in gen:
                count += 1
                if count >= 100:
                    break


Calling async from threads
--------------------------

``aiomisc.sync_await``
++++++++++++++++++++++

Execute an async function synchronously from a thread. Automatically detects
the running event loop:

.. code-block:: python

    import asyncio
    import aiomisc


    async def async_operation():
        await asyncio.sleep(0.1)
        return "async result"


    @aiomisc.threaded
    def thread_function():
        # Call async function from thread
        result = aiomisc.sync_await(async_operation())
        return f"got: {result}"


    async def main():
        result = await thread_function()
        print(result)  # got: async result


    with aiomisc.entrypoint() as loop:
        loop.run_until_complete(main())


``aiomisc.sync_wait_coroutine``
+++++++++++++++++++++++++++++++

Lower-level function that requires explicit loop argument:

.. code-block:: python

    import asyncio
    import aiomisc


    async def fetch_data():
        await asyncio.sleep(0.1)
        return {"key": "value"}


    @aiomisc.threaded
    def process_in_thread(loop):
        # Explicit loop argument
        data = aiomisc.sync_wait_coroutine(loop, fetch_data())
        return data["key"]


    with aiomisc.entrypoint() as loop:
        result = loop.run_until_complete(process_in_thread(loop))
        print(result)  # value


``aiomisc.FromThreadChannel``
-----------------------------

A channel for sending data from threads to async code. Unlike
``IteratorWrapper``, you control when and what to send:

.. code-block:: python

    import asyncio
    import aiomisc


    async def consumer(channel: aiomisc.FromThreadChannel):
        async for item in channel:
            print(f"Received: {item}")


    @aiomisc.threaded_separate
    def producer(channel: aiomisc.FromThreadChannel):
        for i in range(5):
            channel.put(i)  # Send to async consumer
        channel.close()  # Signal end of data


    async def main():
        channel = aiomisc.FromThreadChannel(max_size=2)

        await asyncio.gather(
            consumer(channel),
            producer(channel),
        )


    asyncio.run(main())


``contextvars`` support
-----------------------

All decorators automatically copy context variables to the thread:

.. code-block:: python

    import asyncio
    import contextvars
    import aiomisc


    request_id = contextvars.ContextVar("request_id")


    @aiomisc.threaded
    def log_with_context(message):
        print(f"[{request_id.get()}] {message}")


    async def handle_request(req_id):
        request_id.set(req_id)
        await log_with_context("Processing request")


    async def main():
        await asyncio.gather(
            handle_request("req-001"),
            handle_request("req-002"),
            handle_request("req-003"),
        )


    asyncio.run(main())


Output::

    [req-001] Processing request
    [req-002] Processing request
    [req-003] Processing request


.. note::

    Context variables are copied to threads, so modifications in threads
    don't affect the parent context.


``aiomisc.ThreadPoolExecutor``
------------------------------

A fast thread pool implementation used by aiomisc internally.

Manual setup:

.. code-block:: python

    import asyncio
    from aiomisc import ThreadPoolExecutor

    loop = asyncio.get_event_loop()
    thread_pool = ThreadPoolExecutor(4)
    loop.set_default_executor(thread_pool)


.. note::

    The ``entrypoint`` context manager sets this automatically.
    Use the ``pool_size`` argument to control thread pool size:

    .. code-block:: python

        with aiomisc.entrypoint(pool_size=8) as loop:
            ...


---

``@aiomisc.timeout``
====================

Decorator that ensures the execution time limit for the decorated
function is met.

.. code-block:: python

    from aiomisc import timeout

    @timeout(1)
    async def bad_func():
        await asyncio.sleep(2)


What happens on timeout
+++++++++++++++++++++++

When the decorated function exceeds the specified time limit,
an ``asyncio.TimeoutError`` exception is raised. The underlying
coroutine is cancelled automatically.

.. code-block:: python

    import asyncio
    from aiomisc import timeout

    @timeout(1)
    async def slow_operation():
        await asyncio.sleep(10)
        return "completed"

    async def main():
        try:
            result = await slow_operation()
        except asyncio.TimeoutError:
            print("Operation timed out!")

    asyncio.run(main())


Handling TimeoutError
+++++++++++++++++++++

You should always handle ``asyncio.TimeoutError`` when calling
functions decorated with ``@timeout``, especially in production code:

.. code-block:: python

    import asyncio
    from aiomisc import timeout

    @timeout(5)
    async def fetch_data():
        # Simulate network request
        await asyncio.sleep(10)
        return {"data": "value"}

    async def main():
        try:
            data = await fetch_data()
            print(f"Got data: {data}")
        except asyncio.TimeoutError:
            print("Request timed out, using default value")
            data = {"data": "default"}

    asyncio.run(main())


Comparison with asyncio.wait_for
++++++++++++++++++++++++++++++++

The ``@timeout`` decorator provides similar functionality to
``asyncio.wait_for``, but with a cleaner decorator-based syntax:

+-----------------------------------+-----------------------------------+
| asyncio.wait_for                  | @aiomisc.timeout                  |
+===================================+===================================+
| .. code-block:: python            | .. code-block:: python            |
|                                   |                                   |
|     async def fetch():            |     @timeout(5)                   |
|         await asyncio.sleep(10)   |     async def fetch():            |
|                                   |         await asyncio.sleep(10)   |
|     # At call site:               |                                   |
|     await asyncio.wait_for(       |     # At call site:               |
|         fetch(), timeout=5        |     await fetch()                 |
|     )                             |                                   |
+-----------------------------------+-----------------------------------+

The decorator approach is useful when:

* You want to enforce a timeout for all calls to a function
* The timeout is a property of the function itself, not the caller
* You want cleaner call-site code


Examples with different values
++++++++++++++++++++++++++++++

The timeout value is specified in seconds and can be an integer or float:

.. code-block:: python

    from aiomisc import timeout

    # 100 milliseconds timeout
    @timeout(0.1)
    async def quick_check():
        await asyncio.sleep(0.05)
        return True

    # 30 seconds timeout for long operations
    @timeout(30)
    async def long_operation():
        await asyncio.sleep(25)
        return "done"

    # 2.5 seconds timeout
    @timeout(2.5)
    async def medium_operation():
        await asyncio.sleep(2)
        return "completed"


---

Abstract connection pool
========================

``aiomisc.PoolBase`` is an abstract class for implementing user-defined
connection pool.


Example for ``aioredis``:

.. code-block:: python

    import asyncio
    import aioredis
    import aiomisc


    class RedisPool(aiomisc.PoolBase):
        def __init__(self, uri, maxsize=10, recycle=60):
            super().__init__(maxsize=maxsize, recycle=recycle)
            self.uri = uri

        async def _create_instance(self):
            return await aioredis.create_redis(self.uri)

        async def _destroy_instance(self, instance: aioredis.Redis):
            instance.close()
            await instance.wait_closed()

        async def _check_instance(self, instance: aioredis.Redis):
            try:
                await asyncio.wait_for(instance.ping(1), timeout=0.5)
            except:
                return False

            return True


    async def main():
        pool = RedisPool("redis://localhost")
        async with pool.acquire() as connection:
            await connection.set("foo", "bar")

        async with pool.acquire() as connection:
            print(await connection.get("foo"))


    asyncio.run(main())


---

``WorkerPool``
==============

Python has the ``multiprocessing`` module with ``Pool`` class which
implements a similar worker pool. The IPC in this case uses a completely
synchronous communication method. This module reimplements the process-based
worker pool but IPC is completely asynchronous on the caller side,
meanwhile, workers in separate processes aren't asynchronous.

Example
+++++++

This would be useful when you want to process the data in a separate process,
and the input and output data are not very large. Otherwise, it will work
fine, of course, but you would have to spend time transmitting the data
over IPC.

A good example is parallel image processing. Of course, you can transfer bytes
of images through the IPC of the working pool, but in general, case passing the
file name to the worker will be better. The exception will be cases when the
image payload is significantly smaller the 1KB for example.

Let's write a program that accepts images in JPEG format and creates thumbnails
for this. In this case, you have a file with the original image, and you should
generate the output path for the 'thumbnail' function.

.. note::

    You have to install the Pillow image processing library to run this example.

    Installing pillow with pip:

    .. code-block::

        pip install Pillow

.. code-block:: python

    import asyncio
    import sys
    from multiprocessing import cpu_count
    from typing import Tuple
    from pathlib import Path
    from PIL import Image
    from aiomisc import entrypoint, WorkerPool


    def thumbnail(src_path: str, dest_path: str, box: Tuple[int, int]):
        img = Image.open(src_path)
        img.thumbnail(box)
        img.save(
            dest_path, "JPEG", quality=65,
            optimize=True,
            icc_profile=img.info.get('icc_profile'),
            exif=img.info.get('exif'),
        )
        return img.size


    sizes = [
        (1024, 1024),
        (512, 512),
        (256, 256),
        (128, 128),
        (64, 64),
        (32, 32),
    ]


    async def amain(path: Path):
        # Create directories
        for size in sizes:
            size_dir = "x".join(map(str, size))
            size_path = (path / 'thumbnails' / size_dir)
            size_path.mkdir(parents=True, exist_ok=True)

        # Create and run WorkerPool
        async with WorkerPool(cpu_count()) as pool:
            tasks = []
            for image in path.iterdir():
                if not image.name.endswith(".jpg"):
                    continue

                if image.is_relative_to(path / 'thumbnails'):
                    continue

                for size in sizes:
                    rel_path = image.relative_to(path).parent
                    size_dir = "x".join(map(str, size))
                    dest_path = (
                        path / rel_path /
                        'thumbnails' / size_dir /
                        image.name
                    )

                    tasks.append(
                        pool.create_task(
                            thumbnail,
                            str(image),
                            str(dest_path),
                            size
                        )
                    )

            await asyncio.gather(*tasks)


    if __name__ == '__main__':
        with entrypoint() as loop:
            image_dir = Path(sys.argv[1])
            loop.run_until_complete(amain(image_dir))


This example takes the image directory as the first command-line argument and
creates directories for the thumbnails. After that, a ``WorkerPool`` is started
with as many processes as the processor has cores.

The main process creates tasks for the workers, each task is a conversion of
one file to one size, after which all tasks fall into the ``WorkerPool``
instance.

The ``WorkerPool`` processes the tasks concurrently, but only one job for one
a worker at the same time.


---

``ProcessPoolExecutor``
=======================

This is a simple process pool executor implementation.

Example:

.. code-block:: python

    import asyncio
    import time
    import os
    from aiomisc import ProcessPoolExecutor

    def process_inner():
        for _ in range(10):
            print(os.getpid())
            time.sleep(1)

        return os.getpid()


    loop = asyncio.get_event_loop()
    process_pool = ProcessPoolExecutor(4)


    async def main():
        print(
            await asyncio.gather(
                loop.run_in_executor(process_pool, process_inner),
                loop.run_in_executor(process_pool, process_inner),
                loop.run_in_executor(process_pool, process_inner),
                loop.run_in_executor(process_pool, process_inner),
            )
        )

    loop.run_until_complete(main())


---

Circuit Breaker
===============

`Circuit breaker is a design pattern`_ used in software development.
It is used to detect failures and encapsulates the logic of preventing a
failure from constantly recurring, during maintenance, temporary external
system failure, or unexpected system difficulties.

The following example demonstrates the simple usage of the current
implementation of ``aiomisc.CircuitBreaker``.
An instance of ``CircuitBreaker`` collecting function call statistics.
That contains counters mapping with successful and failed function calls.
Function calls must be wrapped by the `CircuitBreaker.call`
the method to gather it.

Usage example:

.. code-block:: python
    :name: test_circuit_breaker

    from aiohttp import web, ClientSession
    from aiomisc.service.aiohttp import AIOHTTPService
    import aiohttp
    import aiomisc


    async def public_gists(request):
        async with aiohttp.ClientSession() as session:
            # Using as context manager
            with request.app["github_cb"].context():
                url = 'https://api.github.com/gists/public'
                async with session.get(url) as response:
                    data = await response.text()

        return web.Response(
            text=data,
            headers={"Content-Type": "application/json"}
        )


    class API(AIOHTTPService):
        async def create_application(self):
            app = web.Application()
            app.add_routes([web.get('/', public_gists)])

            # When 30% errors in 20 seconds
            # Will be broken for 5 seconds
            app["github_cb"] = aiomisc.CircuitBreaker(
                error_ratio=0.2,
                response_time=20,
                exceptions=[aiohttp.ClientError],
                broken_time=5
            )

            return app


    async def main():
        async with ClientSession() as session:
            async with session.get("http://localhost:8080/") as response:
                assert response.headers


    if __name__ == '__main__':
        with aiomisc.entrypoint(API(port=8080)) as loop:
            loop.run_until_complete(main())


.. _Circuit breaker is a design pattern: http://bit.ly/aimcbwiki


The `CircuitBreaker` object might be one of three states:

    * **PASSING**
    * **BROKEN**
    * **RECOVERING**

.. image:: /_static/cb-states.svg

**PASSING** means all calls will be passed as is and statistics will be gathered.
The next state will be determined after collecting statistics for
``passing_time`` seconds. If an effective error ratio is greater
than `error_ratio` then the next state will be set to **BROKEN**, otherwise,
it will remain unchanged.

**BROKEN** means the wrapped function won't be called and ``CircuitBroken``
an exception will be raised instead. **BROKEN** state will be kept
for ``broken_time`` seconds.

.. note::

    ``CircuitBroken`` exception is a consequence of **BROKEN** or **RECOVERY**
    state and never be accounted for in the statistic.

After that, it changes to **RECOVERING** state. While in that state, a small sample
of the wrapped function calls will be executed and statistics will be
gathered. If the effective error ratio after ``recovery_time`` is lower than
``error_ratio`` then the next state will be set to **PASSING**, and
otherwise - to **BROKEN**.

Argument ``exception_inspector`` is a function that is called whenever
an exception from the list of monitored exceptions occurs. When ``False``
will be returned, this exception will be ignored.

.. image:: /_static/cb-flow.svg


cutout
======

Decorator for ``CircuitBreaker`` which wrapping functions.

.. code-block:: python
    :name: test_cutout

    from aiohttp import web, ClientSession
    from aiomisc.service.aiohttp import AIOHTTPService
    import aiohttp
    import aiomisc


    # When 20% errors in 30 seconds
    # Will be broken on 30 seconds
    @aiomisc.cutout(0.2, 30, aiohttp.ClientError)
    async def fetch(session, url):
        async with session.get(url) as response:
            return await response.text()


    async def public_gists(request):
        async with aiohttp.ClientSession() as session:
            data = await fetch(
                session,
                'https://api.github.com/gists/public'
            )

        return web.Response(
            text=data,
            headers={"Content-Type": "application/json"}
        )


    class API(AIOHTTPService):
        async def create_application(self):
            app = web.Application()
            app.add_routes([web.get('/', public_gists)])
            return app


    async def main():
        async with ClientSession() as session:
            async with session.get("http://localhost:8080/") as response:
                assert response.headers


    if __name__ == '__main__':
        with aiomisc.entrypoint(API(port=8080)) as loop:
            loop.run_until_complete(main())


---

``@aiomisc.asyncbackoff``
=========================

``asyncbackoff`` it's a decorator that helps you guarantee maximal async
function execution and retrying policy.

The main principle might be described in five rules:

* function will be canceled when executed longer than
  ``deadline`` (if specified)
* function will be canceled when executed longer than
  ``attempt_timeout`` (if specified) and will be retried
* Reattempts performs after ``pause`` seconds (if specified, default is ``0``)
* Reattempts will be performed not more than ``max_tries`` times.
* ``giveup`` argument is a function that decides should give
  up the reattempts or continue retrying.

All these rules work at the same time.

Arguments description:

* ``attempt_timeout`` is maximum execution time for one execution attempt.
* ``deadline`` is maximum execution time for all execution attempts.
* ``pause`` is the time gap between execution attempts.
* ``exceptions`` retrying when these exceptions were raised.
* ``giveup`` (keyword only) is a predicate function that can decide by a given exception if we should continue to do retries.
* ``max_tries`` (keyword only) is maximum count of execution attempts (>= 1).

Decorator that ensures that ``attempt_timeout`` and ``deadline`` time
limits are met by decorated function.

In case of exception, the function will be called again with similar arguments after
``pause`` seconds.


Position arguments notation:

.. code-block:: python

    import aiomisc

    attempt_timeout = 0.1
    deadline = 1
    pause = 0.1

    @aiomisc.asyncbackoff(attempt_timeout, deadline, pause)
    async def db_fetch():
        ...


    @aiomisc.asyncbackoff(0.1, 1, 0.1)
    async def db_save(data: dict):
        ...


    # Passing exceptions for handling
    @aiomisc.asyncbackoff(0.1, 1, 0.1, TypeError, RuntimeError, ValueError)
    async def db_fetch(data: dict):
        ...


Keyword arguments notation:

.. code-block:: python

    import aiomisc

    attempt_timeout = 0.1
    deadline = 1
    pause = 0.1

    @aiomisc.asyncbackoff(attempt_timeout=attempt_timeout,
                          deadline=deadline, pause=pause)
    async def db_fetch():
        ...


    @aiomisc.asyncbackoff(attempt_timeout=0.1, deadline=1, pause=0.1)
    async def db_save(data: dict):
        ...


    # Passing exceptions for handling
    @aiomisc.asyncbackoff(attempt_timeout=0.1, deadline=1, pause=0.1,
                          exceptions=[TypeError, RuntimeError, ValueError])
    async def db_fetch(data: dict):
        ...


    # Will be retried no more than 2 times (3 tries total)
    @aiomisc.asyncbackoff(attempt_timeout=0.5, deadline=1,
                          pause=0.1, max_tries=3,
                          exceptions=[TypeError, RuntimeError, ValueError])
    async def db_fetch(data: dict):
        ...


    # Will be retried only on connection abort (on POSIX systems)
    @asyncbackoff(attempt_timeout=0.5, deadline=1, pause=0.1,
                  exceptions=[OSError],
                  giveup=lambda e: e.errno != errno.ECONNABORTED)
    async def db_fetch(data: dict):
        ...


    # Class based example
    backoff = aiomisc.Backoff(
        attempt_timeout=0.5,
        deadline=1,
        pause=0.1,
        exceptions=[OSError],
        giveup=lambda e: e.errno != errno.ECONNABORTED
    )

    async def fetcher(data: dict):
        items = await backoff.execute(db_fetch, data)
        # wrap `db_save` with backoff with the same parameters
        await asyncio.gather(*backoff.execute(db_save, item) for item in items)


.. _asyncretry:

``asyncretry``
==============

Shortcut of ``asyncbackoff(None, None, 0, *args, **kwargs)``. Just retries
``max_tries`` times.

.. note::

    By default will be retry when any Exception. It's very simple and useful
    in generic cases, but you should specify an exception list when you're wrapped
    functions calling hundreds of times per second, cause you have a risk be
    the reason for denial of service in case your function calls remote service.

.. code-block:: python

    from aiomisc import asyncretry

    @asyncretry(5)
    async def try_download_file(url):
        ...

    @asyncretry(3, exceptions=(ConnectionError,))
    async def get_cluster_lock():
        ...


---

Context
=======

Services can require each other's data. In this case, you should use ``Context``.

``Context`` is a repository associated with the running ``entrypoint``.

``Context``-object will be created when ``entrypoint`` starts and linked
to the running event loop.

Cross-dependent services might await or set each other's data via the context.

For service instances ``self.context`` is available since ``entrypoint``
started. In other cases ``get_context()`` function returns current context.


.. code-block:: python
    :name: test_context

    import asyncio
    from random import random, randint

    from aiomisc import entrypoint, get_context, Service


    class LoggingService(Service):
        async def start(self):
            context = get_context()

            wait_time = await context['wait_time']

            print('Wait time is', wait_time)
            self.start_event.set()

            while True:
                print('Hello from service', self.name)
                await asyncio.sleep(wait_time)


    class RemoteConfiguration(Service):
        async def start(self):
            # querying from remote server
            await asyncio.sleep(random())

            self.context['wait_time'] = randint(1, 5)


    services = (
        LoggingService(name='#1'),
        LoggingService(name='#2'),
        LoggingService(name='#3'),
        RemoteConfiguration()
    )

    with entrypoint(*services) as loop:
        pass


.. note::

    It's not a silver bullet. In base case services can be configured by
    passing kwargs to the service ``__init__`` method.


---

``Signal``
==========

You can register async callback functions for specific events of an entrypoint.

``pre_start``
+++++++++++++

``pre_start`` signal occurs on entrypoint starting up before any service has started.

.. code-block:: python

    from aiomisc import entrypoint, receiver

    @receiver(entrypoint.PRE_START)
    async def prepare_database(entrypoint, services):
      ...

    with entrypoint() as loop:
        loop.run_forever()


``post_start``
++++++++++++++

``post_start`` signal occurs next entrypoint starting up after all services have
been started.

.. code-block:: python

    from aiomisc import entrypoint, receiver

    @receiver(entrypoint.POST_START)
    async def startup_notifier(entrypoint, services):
      ...

    with entrypoint() as loop:
        loop.run_forever()


``pre_stop``
++++++++++++

``pre_stop`` signal occurs on entrypoint shutdown before any service have been
stopped.

.. code-block:: python

    from aiomisc import entrypoint, receiver

    @receiver(entrypoint.PRE_STOP)
    async def shutdown_notifier(entrypoint):
      ...

    with entrypoint() as loop:
        loop.run_forever()


``post_stop``
+++++++++++++

``post_stop`` signal occurs on entrypoint shutdown after all services have been
stopped.

.. code-block:: python

    from aiomisc import entrypoint, receiver

    @receiver(entrypoint.POST_STOP)
    async def cleanup(entrypoint):
      ...

    with entrypoint() as loop:
        loop.run_forever()


---

entrypoint
==========

In the generic case, the entrypoint helper creates an event loop and
cancels already running coroutines on exit.

.. code-block:: python
    :name: test_entrypoint_simple

    import asyncio
    import aiomisc

    async def main():
        await asyncio.sleep(1)

    with aiomisc.entrypoint() as loop:
        loop.run_until_complete(main())


Complete example:

.. code-block:: python
    :name: test_entrypoint_complex

    import asyncio
    import aiomisc
    import logging
    import signal

    async def main():
        await asyncio.sleep(1)
        logging.info("Hello there")

    with aiomisc.entrypoint(
        pool_size=2,
        log_level='info',
        log_format='color',                            # default when "rich" absent
        log_buffer_size=1024,                          # default
        log_flush_interval=0.2,                        # default
        log_config=True,                               # default
        debug=False,                                   # default
        catch_signals=(signal.SIGINT, signal.SIGTERM), # default
        shutdown_timeout=60,                           # default
    ) as loop:
        loop.run_until_complete(main())

Running entrypoint from async code

.. code-block:: python
    :name: test_entrypoint_async

    import asyncio
    import aiomisc
    import logging
    from aiomisc.service.periodic import PeriodicService

    log = logging.getLogger(__name__)

    class MyPeriodicService(PeriodicService):
        async def callback(self):
            log.info('Running periodic callback')
            # ...

    async def main():
        service = MyPeriodicService(interval=1, delay=0)  # once per minute

        # returns an entrypoint instance because event-loop
        # already running and might be get via asyncio.get_event_loop()
        async with aiomisc.entrypoint(service) as ep:
            try:
                await asyncio.wait_for(ep.closing(), timeout=1)
            except asyncio.TimeoutError:
                pass


    asyncio.run(main())

Dynamic running of services
+++++++++++++++++++++++++++

Sometimes it is not enough to add services to the entrypoint at the start,
or it is not possible to get the service parameters before the start of
the event-loop. In this case it is possible to start services after the
event-loop has started, this feature available from version ``17``.

.. code-block:: python

    import asyncio
    import aiomisc
    import logging

    from aiomisc.service.periodic import PeriodicService

    log = logging.getLogger(__name__)


    class MyPeriodicService(PeriodicService):
        async def callback(self):
            log.info('Running periodic callback')


    async def add_services():
        entrypoint = aiomisc.entrypoint.get_current()

        services = [
            MyPeriodicService(interval=2, delay=1),
            MyPeriodicService(interval=2, delay=0),
        ]

        await entrypoint.start_services(*services)
        await asyncio.sleep(10)
        await entrypoint.stop_services(*services)


    with aiomisc.entrypoint() as loop:
        loop.create_task(add_services())
        loop.run_forever()


Configuration from environment
++++++++++++++++++++++++++++++

Module support configuration from environment variables:

* ``AIOMISC_LOG_LEVEL`` - default logging level
* ``AIOMISC_LOG_FORMAT`` - default log format
* ``AIOMISC_LOG_DATE_FORMAT`` - default logging date format
* ``AIOMISC_LOG_CONFIG`` - should logging be configured
* ``AIOMISC_LOG_FLUSH`` - interval between logs flushing from buffer
* ``AIOMISC_LOG_BUFFERING`` - should logging be buffered
* ``AIOMISC_LOG_BUFFER_SIZE`` - maximum log buffer size
* ``AIOMISC_POOL_SIZE`` - thread pool size
* ``AIOMISC_USE_UVLOOP`` - should use uvloop when it available, ``0`` to disable
* ``AIOMISC_SHUTDOWN_TIMEOUT`` - If, after receiving the signal, the program
  does not terminate within this timeout, a force-exit occurs.


``run()`` shortcut
==================

``aiomisc.run()`` - it's the short way to create and destroy
``aiomisc.entrypoint``. It's very similar to ``asyncio.run()``
but handle ``Service``'s and other ``entrypoint``'s kwargs.

.. code-block:: python
    :name: test_ep_run_simple

    import asyncio
    import aiomisc

    async def main():
        loop = asyncio.get_event_loop()
        now = loop.time()
        await asyncio.sleep(0.1)
        assert now < loop.time()


    aiomisc.run(main())

Logging configuration
=====================

``entrypoint`` accepts ``log_format`` argument with a specific set of formats,
in which logs will be written to stderr:

* ``stream`` - Python's default logging handler
* ``color`` - logging with `colorlog` module
* ``json`` - json structure per each line
* ``syslog`` - logging using stdlib `logging.handlers.SysLogHandler`
* ``plain`` - just log messages, without date or level info
* ``journald`` - available only when `logging-journald` module
  has been installed.
* ``rich``/``rich_tb`` - available only when `rich` module has been installed.
  ``rich_tb`` it's the same as ``rich`` but with fully expanded tracebacks.

Additionally, you can specify log level using ``log_level`` argument and date
format using ``log_date_format`` parameters.

An ``entrypoint`` will call ``aiomisc.log.basic_config`` function implicitly
using passed ``log_*`` parameters. Alternatively you can call
``aiomisc.log.basic_config`` function manually passing it already created
eventloop.

However, you can configure logging earlier using ``aiomisc_log.basic_config``,
but you will lose log buffering and flushing in a separate thread.
This function is what is actually called during the logging configuration,
the ``entrypoint`` passes a wrapper for the handler there to flush it into
the separate thread.

.. code-block:: python

    import logging

    from aiomisc_log import basic_config


    basic_config(log_format="color")
    logging.info("Hello")

If you want to configure logging before the ``entrypoint`` is started,
for example after the arguments parsing, it is safe to configure it twice
(or more).

.. code-block:: python

    import logging

    import aiomisc
    from aiomisc_log import basic_config


    basic_config(log_format="color")
    logging.info("Hello from usual python")


    async def main():
        logging.info("Hello from async python")


    with aiomisc.entrypoint(log_format="color") as loop:
        loop.run_until_complete(main())


Sometimes you want to configure logging manually, the following example
demonstrates how to do this:

.. code-block:: python

    import os
    import logging
    from logging.handlers import RotatingFileHandler
    from gzip import GzipFile

    import aiomisc


    class GzipLogFile(GzipFile):
        def write(self, data) -> int:
            if isinstance(data, str):
                data = data.encode()
            return super().write(data)


    class RotatingGzipFileHandler(RotatingFileHandler):
        """ Really added just for example you have to test it properly """

        def shouldRollover(self, record):
            if not os.path.isfile(self.baseFilename):
                return False
            if self.stream is None:
                self.stream = self._open()
            return 0 < self.maxBytes < os.stat(self.baseFilename).st_size

        def _open(self):
            return GzipLogFile(filename=self.baseFilename, mode=self.mode)


    async def main():
        for _ in range(1_000):
            logging.info("Hello world")


    with aiomisc.entrypoint(log_config=False) as loop:
        gzip_handler = RotatingGzipFileHandler(
            "app.log.gz",
            # Maximum 100 files by 10 megabytes
            maxBytes=10 * 2 ** 20, backupCount=100
        )
        stream_handler = logging.StreamHandler()

        formatter = logging.Formatter(
            "[%(asctime)s] <%(levelname)s> "
            "%(filename)s:%(lineno)d (%(threadName)s): %(message)s"
        )

        gzip_handler.setFormatter(formatter)
        stream_handler.setFormatter(formatter)

        logging.basicConfig(
            level=logging.INFO,
            # Wrapping all handlers in separate streams will not block the
            # event-loop even if gzip takes a long time to open the
            # file.
            handlers=map(
                aiomisc.log.wrap_logging_handler,
                (gzip_handler, stream_handler)
            )
        )
        loop.run_until_complete(main())


---

Logging configuration
=====================

Default logging configuration might be configured by setting environment variables:

* `AIOMISC_LOG_LEVEL` - default logging level
* `AIOMISC_LOG_FORMAT` - default log format
* `AIOMISC_LOG_CONFIG` - should logging be configured
* `AIOMISC_LOG_FLUSH` - interval between logs flushing from buffer
* `AIOMISC_LOG_BUFFER` - maximum log buffer size

.. code-block:: shell

    $ export AIOMISC_LOG_LEVEL=debug
    $ export AIOMISC_LOG_FORMAT=rich


Color
+++++

Setting up colorized logs:

.. code-block:: python
    :name: test_logging_color

    import logging
    from aiomisc.log import basic_config


    # Configure logging
    basic_config(level=logging.INFO, buffered=False, log_format='color')

JSON
++++

Setting up json logs:

.. code-block:: python
    :name: test_logging_json

    import logging
    from aiomisc.log import basic_config


    # Configure logging
    basic_config(level=logging.INFO, buffered=False, log_format='json')

JournalD
++++++++

`JournalD`_ daemon for collecting logs. It's a part of the systemd.
`aiomisc.basic_config` has support for using `JournalD`_ for store logs.

.. note::

    This handler is the default when the program starting as a systemd service.

    `aiomisc.log.LogFormat.default()` will returns `journald`  in this case.

.. code-block:: python

    import logging
    from aiomisc.log import basic_config

    # Configure rich log handler
    basic_config(level=logging.INFO, buffered=False, log_format='journald')

    logging.info("JournalD log record")

.. _JournalD: https://www.freedesktop.org/software/systemd/man/systemd-journald.service.html


Rich
++++

`Rich`_ is a Python library for rich text and beautiful formatting in the terminal.

`aiomisc.basic_config` has support for using `Rich`_ as a logging handler.
But it isn't dependency and you have to install `Rich`_ manually.

.. code-block:: bash

    pip install rich

.. note::

    This handler is the default when the `Rich` has been installed.

.. code-block:: python
    :name: test_rich_handlers

    import logging
    from aiomisc.log import basic_config

    # Configure rich log handler
    basic_config(level=logging.INFO, buffered=False, log_format='rich')

    logging.info("Rich logger")

    # Configure rich log handler with rich tracebacks display
    basic_config(level=logging.INFO, buffered=False, log_format='rich_tb')

    try:
        1 / 0
    except:
        logging.exception("Rich traceback logger")

.. _Rich: https://pypi.org/project/rich/

Disabled
++++++++

Disable to configure logging handler. Useful when you want to configure your own logging handlers using
`handlers=` argument.

.. code-block:: python
    :name: test_log_disabled

    import logging
    from aiomisc.log import basic_config

    # Configure your own log handlers
    basic_config(
        level=logging.INFO,
        log_format='disabled',
        handlers=[logging.StreamHandler()],
        buffered=False,
    )

    logging.info("Use default python logger for example")



Buffered log handler
++++++++++++++++++++

Parameter `buffered=True` enables a memory buffer that flushes logs in a thread. In case the `handlers=`
each will be buffered.

.. code-block:: python
    :name: test_logging_buffered

    import asyncio
    import logging
    from aiomisc.log import basic_config
    from aiomisc.periodic import PeriodicCallback


    # Configure logging globally
    basic_config(level=logging.INFO, buffered=False, log_format='json')

    async def write_log():
        logging.info("Hello %f", asyncio.get_event_loop().time())

    async def main():
        # Configure
        basic_config(
            level=logging.INFO,
            buffered=True,
            log_format='color',
            flush_interval=0.5
        )

        periodic = PeriodicCallback(write_log)
        periodic.start(0.3)

        # Wait for flush just for example
        await asyncio.sleep(1)

    if __name__ == '__main__':
        with asyncio.Runner() as runner:
            runner.run(main())


.. note::

    ``entrypoint`` accepts ``log_format`` parameter for configure it.

    List of all supported log formats is available from
    ``aiomisc.log.LogFormat.choices()``


---

Statistic counters
==================

``aiomisc`` contains internal statistic counters. You may read this by
`aiomisc.get_statistics()` function.

Statistic instances create dynamically. You might set custom names for this
using ``statistic_name: Optional[str] = None`` argument for compatible
entities.

.. code-block:: python

    import aiomisc

    async def main():
        for metric in aiomisc.get_statistics():
            print(
                str(metric.kind.__name__),
                metric.name,
                metric.metric,
                metric.value
            )


    with aiomisc.entrypoint() as loop:
        loop.run_until_complete(main())


This code will print something like this:

.. code-block::

    ContextStatistic None get 0
    ContextStatistic None set 0
    ThreadPoolStatistic logger submitted 1
    ThreadPoolStatistic logger sum_time 0
    ThreadPoolStatistic logger threads 1
    ThreadPoolStatistic logger done 0
    ThreadPoolStatistic logger success 0
    ThreadPoolStatistic logger error 0
    ThreadPoolStatistic default submitted 0
    ThreadPoolStatistic default sum_time 0
    ThreadPoolStatistic default threads 12
    ThreadPoolStatistic default done 0
    ThreadPoolStatistic default success 0
    ThreadPoolStatistic default error 0


---

``@aiomisc.aggregate``
======================

Parametric decorator that aggregates multiple
(but no more than ``max_count`` defaulting to ``None``) single-argument
executions (``res1 = await func(arg1)``, ``res2 = await func(arg2)``, ...)
of an asynchronous function with variadic positional arguments
(``async def func(*args, pho=1, bo=2) -> Iterable``) into its single execution
with multiple positional arguments
(``res1, res2, ... = await func(arg1, arg2, ...)``) collected within a time
window ``leeway_ms``. It offers a trade-off between latency and throughput.

If ``func`` raises an exception, then, all of the aggregated calls will
propagate the same exception. If one of the aggregated calls gets canceled
during the ``func`` execution, then, another will try to execute the ``func``.

This decorator may be useful if the ``func`` executes slow IO-tasks,
is frequently called, and using cache is not a good option. As a toy example,
assume that ``func`` fetches a record from the database by user ID and it is
called during each request to our service. If it takes 100 ms to fetch a
record and the load is 1000 RPS, then, with a 10% increase of the delay
(to 110 ms), it may decrease the number of requests to the database by
10 times (to 100 QPS).

.. image:: /_static/aggregate-flow.svg

.. code-block:: python
    :name: test_aggregate

    import asyncio
    import math
    from aiomisc import aggregate, entrypoint

    @aggregate(leeway_ms=10, max_count=2)
    async def pow(*nums: float, power: float = 2.0):
        return [math.pow(num, power) for num in nums]

    async def main():
        await asyncio.gather(pow(1.0), pow(2.0))

    with entrypoint() as loop:
        loop.run_until_complete(main())

To employ a more low-level approach one can use `aggregate_async` instead.
In this case, the aggregating function accepts `Arg` parameters, each containing
`value` and `future` attributes. It is responsible for setting the results
of execution for all the futures (instead of returning values).

.. code-block:: python
    :name: test_aggregate_async

    import asyncio
    import math
    from aiomisc import aggregate_async, entrypoint
    from aiomisc.aggregate import Arg


    @aggregate_async(leeway_ms=10, max_count=2)
    async def pow(*args: Arg, power: float = 2.0):
        for arg in args:
            arg.future.set_result(math.pow(arg.value, power))

    async def main():
        await asyncio.gather(pow(1), pow(2))

    with entrypoint() as loop:
        loop.run_until_complete(main())


---

asynchronous file operations
============================

Asynchronous files operations including support for data compression on the fly.
Based on the thread pool under the hood.

.. code-block:: python
    :name: test_io

    import aiomisc
    import tempfile
    from pathlib import Path


    async def file_write():
        with tempfile.TemporaryDirectory() as tmp:
            fname = Path(tmp) / 'test.txt'

            # Some tools, such as mypy, will not be able to infer the type
            # from the `async_open` function based on the `b` character passed
            # to the mode.
            # We'll have to tell the type here explicitly.
            afp: aiomisc.io.AsyncTextIO

            async with aiomisc.io.async_open(fname, 'w+') as afp:
                await afp.write("Hello")
                await afp.write(" ")
                await afp.write("world")

                await afp.seek(0)
                print(await afp.read())


    with aiomisc.entrypoint() as loop:
        loop.run_until_complete(file_write())


This is the way to working with files based on threads.
It's very similar to `aiofiles`_ project and same limitations.

Of course, you can use `aiofile`_ project for this. But it's not a
silver bullet and has OS API-related limitations.

In general, for light loads, I would advise you to adhere to the following rules:

* If reading and writing small or big chunks from files with random access
  is the main task in your project, use `aiofile`_.
* Otherwise use this module or `aiofiles`_
* If the main task is to read large chunks of files for processing,
  both of the above methods are not optimal cause you will switch
  context each IO operation, it's often suboptimal for file cache
  and you will be lost execution time for context switches. In case
  for thread-based IO executor implementation thread context
  switches cost might be more expensive than IO operation time
  in summary.

  Just try pack all blocking staff in separate functions and
  call it in a thread pool, see the example below:

  .. code-block:: python
     :name: test_io_file_threaded

     import os
     import aiomisc
     import hashlib
     import tempfile
     from pathlib import Path


     @aiomisc.threaded
     def hash_file(filename, chunk_size=65535, hash_func=hashlib.blake2b):
         hasher = hash_func()

         with open(filename, "rb") as fp:
             for chunk in iter(lambda: fp.read(chunk_size), b""):
                hasher.update(chunk)

         return hasher.hexdigest()


     @aiomisc.threaded
     def fill_random_file(filename, size, chunk_size=65535):
         with open(filename, "wb") as fp:
            while fp.tell() < size:
                fp.write(os.urandom(chunk_size))

            return fp.tell()


     async def main(path):
         filename = path / "one"
         await fill_random_file(filename, 1024 * 1024)
         first_hash = await hash_file(filename)

         filename = path / "two"
         await fill_random_file(filename, 1024 * 1024)
         second_hash = await hash_file(filename)

         assert first_hash != second_hash


     with tempfile.TemporaryDirectory(prefix="random.") as path:
        aiomisc.run(
            main(Path(path))
        )

In the fly compression
----------------------

To enable compression, you need to pass the `compression` argument to the
`async_open` function.

Supported compressors:

* :class:`aiomisc.io.Compression.NONE`
* :class:`aiomisc.io.Compression.GZIP`
* :class:`aiomisc.io.Compression.BZ2`
* :class:`aiomisc.io.Compression.LZMA`

An example of usage:

.. code-block:: python
    :name: test_compressed_gzip_io

    import tempfile
    from aiomisc import run
    from aiomisc.io import async_open, Compression
    from pathlib import Path


    async def file_write():
        with tempfile.TemporaryDirectory() as tmp:
            fname = Path(tmp) / 'test.txt'

            async with async_open(
                fname, 'w+', compression=Compression.GZIP
            ) as afp:
                for _ in range(10000):
                    await afp.write("Hello World\n")

            file_size = fname.stat().st_size
            assert file_size < 10000, f"File too large {file_size} bytes"

    run(file_write())


.. _aiofiles: https://pypi.org/project/aiofiles/
.. _aiofile: https://pypi.org/project/aiofile/


---

Utilities
=========

``select``
++++++++++

In some cases, you should wait for only one of multiple tasks. ``select``
waits first passed awaitable object and returns the list of results.

.. code-block:: python

    import asyncio
    import aiomisc


    async def main():
        loop = asyncio.get_event_loop()
        event = asyncio.Event()
        future = asyncio.Future()

        loop.call_soon(event.set)

        await aiomisc.select(event.wait(), future)
        print(event.is_set())       # True

        event = asyncio.Event()
        future = asyncio.Future()

        loop.call_soon(future.set_result, True)

        results = await aiomisc.select(future, event.wait())
        future_result, event_result = results

        print(results.result())             # True
        print(results.result_idx)           # 0
        print(event_result, future_result)  # None, True


    with aiomisc.entrypoint() as loop:
        loop.run_until_complete(main())


.. warning::

    When you don't want to cancel pending tasks pass ``cancel=False`` argument.
    In this case, you have to handle task completion manually or get warnings.


``cancel_tasks``
++++++++++++++++

All passed tasks will be canceled and the task will be returned:

.. code-block:: python

    import asyncio
    from aiomisc import cancel_tasks


    async def main():
        done, pending = await asyncio.wait([
            asyncio.sleep(i) for i in range(10)
        ], timeout=5)

        print("Done", len(done), "tasks")
        print("Pending", len(pending), "tasks")
        await cancel_tasks(pending)


    asyncio.run(main())


``awaitable``
+++++++++++++

Decorator wraps function and returns a function that returns awaitable object.
In case a function returns a future, the original future will be returned.
In case then the function returns a coroutine, the original coroutine will
be returned. In case than function returns a non-awaitable object, it's will
be wrapped to a new coroutine that just returns this object. It's useful
when you don't want to check function results before
use it in ``await`` expression.

.. code-block:: python

    import asyncio
    import aiomisc


    async def do_callback(func, *args):
        awaitable_func = aiomisc.awaitable(func)

        return await awaitable_func(*args)


    print(asyncio.run(do_callback(asyncio.sleep, 2)))
    print(asyncio.run(do_callback(lambda: 45)))

``bind_socket``
+++++++++++++++

Bind socket and set ``setblocking(False)`` for just created socket.
This detects ``address`` format and selects the socket family automatically.

.. code-block:: python

    from aiomisc import bind_socket

    # IPv4 socket
    sock = bind_socket(address="127.0.0.1", port=1234)

    # IPv6 socket (on Linux IPv4 socket will be bind too)
    sock = bind_socket(address="::1", port=1234)


``RecurringCallback``
+++++++++++++++++++++

Runs coroutine function periodically with user-defined strategy.

.. code-block:: python

    from typing import Union
    from aiomisc import new_event_loop, RecurringCallback


    async def callback():
        print("Hello")


    FIRST_CALL = False


    async def strategy(_: RecurringCallback) -> Union[int, float]:
        global FIRST_CALL
        if not FIRST_CALL:
            FIRST_CALL = True
            # Delay 5 second if just started
            return 5

        # Delay 10 seconds if it is not a first call
        return 10


    if __name__ == '__main__':
        loop = new_event_loop()

        periodic = RecurringCallback(callback)

        task = periodic.start(strategy)
        loop.run_forever()


The main purpose is this class is to provide ability to specify the
asynchronous strategy function, which can be written very flexible.

Also, with the special exceptions, you can control the behavior of the started
``RecurringCallback``.

.. code-block:: python

    from aiomisc import (
        new_event_loop, RecurringCallback, StrategySkip, StrategyStop
    )


    async def strategy(_: RecurringCallback) -> Union[int, float]:
        ...

        # Skip this attempt and wait 10 seconds
        raise StrategySkip(10)

        ...

        # Stop execution
        raise StrategyStop()

if the strategy function returns an incorrect value (not a number), or does
not raise special exceptions, the recurring execution is terminated.


``PeriodicCallback``
++++++++++++++++++++

Runs coroutine function periodically with an optional delay of the first
execution. Uses ``RecurringCallback`` under the hood.

.. code-block:: python

    import asyncio
    import time
    from aiomisc import new_event_loop, PeriodicCallback


    async def periodic_function():
        print("Hello")


    if __name__ == '__main__':
        loop = new_event_loop()

        periodic = PeriodicCallback(periodic_function)

        # Wait 10 seconds and call it each second after that
        periodic.start(1, delay=10)

        loop.run_forever()


``CronCallback``
++++++++++++++++

.. warning::

    You have to install ``croniter`` package for use this feature:

    .. code-block:: shell

        pip install croniter

    Or add extras when installing aiomisc:

    .. code-block:: shell

        pip install aiomisc[cron]



Runs coroutine function with cron scheduling execution.
Uses ``RecurringCallback`` under the hood.

.. code-block:: python

    import asyncio
    import time
    from aiomisc import new_event_loop, CronCallback


    async def cron_function():
        print("Hello")


    if __name__ == '__main__':
        loop = new_event_loop()

        periodic = CronCallback(cron_function)

        # call it each second after that
        periodic.start(spec="* * * * * *")

        loop.run_forever()


---

Services
========

``Services`` is an abstraction to help organize lots of different
tasks in one process. Each service must implement ``start()`` method and can
implement ``stop()`` method.

Service instance should be passed to the ``entrypoint``, and will be started
after the event loop has been created.

.. note::

   Current event-loop will be set before ``start()`` method called.
   The event loop will be set as current for this thread.

   Please avoid using ``asyncio.get_event_loop()`` explicitly inside
   ``start()`` method. Use ``self.loop`` instead:

   .. code-block:: python
      :name: test_service_start_event

      import asyncio
      from threading import Event
      from aiomisc import entrypoint, Service

      event = Event()

      class MyService(Service):
        async def start(self):
            # Send signal to entrypoint for continue running
            self.start_event.set()

            event.set()
            # Start service task
            await asyncio.sleep(3600)


      with entrypoint(MyService()) as loop:
          assert event.is_set()


Method ``start()`` creates as a separate task that can run forever. But in
this case ``self.start_event.set()`` should be called for notifying
``entrypoint``.

During graceful shutdown method ``stop()`` will be called first,
and after that, all running tasks will be canceled (including ``start()``).


This package contains some useful base classes for simple services writing.

.. toctree::
   :maxdepth: 2

   network
   periodic
   configuration
   web
   grpc
   system


---

API Reference
=============

.. toctree::
   :glob:
   :numbered:
   :maxdepth: 3

   aiomisc
   aiomisc_log
   aiomisc_worker


---

Pytest plugin
=============

.. _aiomisc-pytest: https://pypi.org/project/aiomisc-pytest

All pytest plugin code lives in the ``aiomisc`` package itself.
The aiomisc-pytest_ package is required only for plugin registration
(it contains the pytest entry point). Install both:

.. code-block:: bash

    pip install aiomisc aiomisc-pytest

Basic usage
-----------

The plugin automatically provides an event loop and an entrypoint for every
async test. Simply write ``async def test_...`` functions:

.. code-block:: python

    import asyncio

    async def test_sample(event_loop: asyncio.AbstractEventLoop):
        f = event_loop.create_future()
        event_loop.call_soon(f.set_result, True)
        assert await f

Async fixtures work the same way:

.. code-block:: python

    import asyncio
    import pytest


    @pytest.fixture
    async def my_fixture():
        await asyncio.sleep(0)
        yield "value"


    async def test_with_fixture(my_fixture):
        assert my_fixture == "value"

Fixtures
--------

The plugin provides the following fixtures:

Core fixtures
+++++++++++++

``entrypoint``
    **Scope:** function, autouse

    Creates an ``aiomisc.Entrypoint`` instance using the current
    ``event_loop``. Starts all services from the ``services`` fixture and
    populates the context from ``default_context``.

``localhost``
    **Scope:** session

    Returns the localhost address (``"127.0.0.1"`` or ``"[::1]"`` depending
    on what is available on the system).

``loop_debug``
    **Scope:** session

    Returns the value of ``--aiomisc-debug`` command-line option.

``aiomisc_test_timeout``
    **Scope:** session

    Returns the value of ``--aiomisc-test-timeout`` command-line option.

``thread_pool_size``
    **Scope:** session

    Returns the value of ``--aiomisc-pool-size`` command-line option.

``tcp_proxy``
    **Scope:** session

    Returns the ``TCPProxy`` class for emulating network problems.
    See `TCPProxy`_ section below.

Overridable fixtures
++++++++++++++++++++

These fixtures have sensible defaults but can be overridden in your
``conftest.py`` or test module to change behavior or scope.

``event_loop``
    **Scope:** function, autouse, **Default:** creates a new event loop per test

    Creates and manages an asyncio event loop. Configures thread pool,
    debug mode, and exception handling. Closes the loop on teardown.

    Override this fixture to change its scope (module or session) when you
    need async fixtures that outlive a single test. See
    `Scoped event loops and fixtures`_ for details.

``services``
    **Scope:** function, **Default:** empty list

    Return a list of ``aiomisc.Service`` instances to start inside
    the entrypoint.

    .. code-block:: python

        from typing import Iterable

        import aiomisc
        import pytest


        class MyService(aiomisc.Service):
            async def start(self) -> None:
                self.start_event.set()
                await asyncio.sleep(3600)


        @pytest.fixture
        def services() -> Iterable[aiomisc.Service]:
            return [MyService()]

``default_context``
    **Scope:** function, **Default:** empty dict

    Return a mapping of values to populate in the entrypoint context.

    .. code-block:: python

        from typing import Any, Mapping

        import pytest


        @pytest.fixture
        def default_context() -> Mapping[str, Any]:
            return {
                "foo": "bar",
                "bar": "foo",
            }

``entrypoint_kwargs``
    **Scope:** function, **Default:** ``{"log_config": False}``

    Return extra keyword arguments for the ``entrypoint()`` constructor.

    .. code-block:: python

        import pytest


        @pytest.fixture
        def entrypoint_kwargs() -> dict:
            return dict(log_config=False)

``thread_pool_executor``
    **Scope:** function, **Default:** ``aiomisc.ThreadPoolExecutor``

    Return the thread pool executor class to use.

    .. code-block:: python

        import concurrent.futures
        import pytest


        @pytest.fixture
        def thread_pool_executor() -> type[concurrent.futures.ThreadPoolExecutor]:
            return concurrent.futures.ThreadPoolExecutor

Port and socket fixtures
++++++++++++++++++++++++

``aiomisc_unused_port_factory``
    **Scope:** function

    A callable factory that returns an unused port on each call.
    Sockets are cleaned up after test teardown.

``aiomisc_unused_port``
    **Scope:** function

    A single unused port number (uses ``aiomisc_unused_port_factory``
    internally).

``aiomisc_socket_factory``
    **Scope:** function

    A callable factory that returns a ``PortSocket(port, socket)``
    named tuple with a bound socket.

Markers
-------

``@pytest.mark.catch_loop_exceptions``
    Uncaught event loop exceptions will fail the test.

    .. code-block:: python

        import asyncio
        import pytest


        @pytest.mark.catch_loop_exceptions
        async def test_with_errors(event_loop):
            async def fail():
                await asyncio.sleep(0)
                raise Exception()

            event_loop.create_task(fail())
            await asyncio.sleep(0.1)

``@pytest.mark.forbid_get_event_loop``
    Forbids calling ``asyncio.get_event_loop()`` during the test.

    .. code-block:: python

        import asyncio
        import pytest


        @pytest.mark.forbid_get_event_loop
        async def test_no_get_loop():
            def bad():
                asyncio.get_event_loop()

            with pytest.raises(Exception):
                bad()

Command-line options
--------------------

``--aiomisc-debug``
    Enable event loop debug mode. Default: ``False``.

``--aiomisc-pool-size``
    Thread pool size. Default: ``4``.

``--aiomisc-test-timeout``
    Per-test timeout in seconds. Default: ``None`` (no timeout).

Environment variables
---------------------

``AIOMISC_USE_UVLOOP``
    Set to ``"0"``, ``"no"``, or ``"false"`` to disable uvloop.
    Default: ``"1"`` (enabled if uvloop is installed).

``AIOMISC_LOOP_AUTOUSE``
    Set to ``"0"`` to disable autouse on ``event_loop`` and ``entrypoint``
    fixtures. Default: ``"1"``.

Scoped event loops and fixtures
--------------------------------

Why change the scope?
+++++++++++++++++++++

By default, ``event_loop`` and ``entrypoint`` are **function-scoped**: a
fresh event loop is created for every test and closed on teardown. This is
the safest default because each test gets a clean state.

However, some async resources are expensive to create and should be shared
across tests: database connection pools, HTTP client sessions, preloaded
caches, and so on. In pytest, you express this by giving the fixture a
wider scope (``"module"`` or ``"session"``).

The problem is that a wider-scoped async fixture **must run on an event loop
that lives at least as long as the fixture itself**. If the fixture is
session-scoped but the event loop is function-scoped, the loop will be
closed after the first test, and the fixture's teardown (and every
subsequent test that uses it) will fail.

The solution: override ``event_loop`` with the same scope as your
widest async fixture.

The key rule
++++++++++++

.. important::

    The ``event_loop`` fixture scope must be **greater than or equal to** the
    scope of every async fixture that depends on it.

    ``session >= module >= class >= function``

    For example, if you have a ``scope="module"`` async fixture, you need at
    least a ``scope="module"`` event loop.

Module-scoped loop
++++++++++++++++++

Override ``event_loop`` in your test module or in a ``conftest.py`` that
applies to the relevant directory:

.. code-block:: python

    import asyncio
    from typing import Iterator, AsyncIterator
    import pytest


    @pytest.fixture(scope="module")
    def event_loop() -> Iterator[asyncio.AbstractEventLoop]:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            yield loop
        finally:
            loop.close()
            asyncio.set_event_loop(None)


    @pytest.fixture(scope="module")
    async def shared_resource() -> AsyncIterator[Resource]:
        resource = await create_expensive_resource()
        yield resource
        await resource.close()


    async def test_first(shared_resource):
        assert shared_resource is not None


    async def test_second(shared_resource):
        assert shared_resource is not None

All tests in the module share the same event loop and the same
``shared_resource`` instance. The resource is created once before the
first test in the module and torn down after the last one.

Session-scoped loop
+++++++++++++++++++

For fixtures that must survive the entire test session, place a
session-scoped ``event_loop`` override in a ``conftest.py``. Because this
affects every test that inherits it, it is recommended to put it in a
subdirectory ``conftest.py`` rather than the root one, so that only the
tests that need it are affected.

.. code-block:: bash

    tests/
        conftest.py               # root conftest (default fixtures)
        test_unit.py              # uses default function-scoped loop
        integration/
            conftest.py           # session-scoped event_loop override
            test_database.py      # shares the session loop
            test_api.py           # shares the session loop

.. code-block:: python

    # tests/integration/conftest.py
    import asyncio
    from typing import Iterator
    import pytest


    @pytest.fixture(scope="session")
    def event_loop() -> Iterator[asyncio.AbstractEventLoop]:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            yield loop
        finally:
            loop.close()
            asyncio.set_event_loop(None)

Then use session-scoped async fixtures as usual:

.. code-block:: python

    # tests/integration/conftest.py (continued)
    from typing import AsyncIterator
    import pytest


    @pytest.fixture(scope="session")
    async def db_pool() -> AsyncIterator[Pool]:
        pool = await create_pool(dsn="postgresql://localhost/test")
        yield pool
        await pool.close()

.. code-block:: python

    # tests/integration/test_database.py
    async def test_insert(db_pool):
        async with db_pool.acquire() as conn:
            await conn.execute("INSERT INTO t VALUES (1)")


    async def test_select(db_pool):
        async with db_pool.acquire() as conn:
            row = await conn.fetchrow("SELECT 1 AS n")
            assert row["n"] == 1

Session-scoped async generator fixtures
++++++++++++++++++++++++++++++++++++++++

Async generator fixtures (those that use ``yield``) require extra care.
When the scope is wider than the ``entrypoint`` fixture, the entrypoint
must not destroy the generator during its own teardown.

This works correctly: the ``entrypoint`` is function-scoped and does not
own the session-scoped event loop, so it will not call
``shutdown_asyncgens`` and the generator survives between tests.

.. code-block:: python

    from typing import AsyncIterator
    import pytest


    async def some_agen() -> AsyncIterator[int]:
        for i in range(100):
            yield i + 1


    @pytest.fixture(scope="session")
    async def async_gen_fixture() -> AsyncIterator[int]:
        agen = some_agen()
        val = await agen.__anext__()
        assert val == 1
        val = await agen.__anext__()
        assert val == 2
        yield val
        # teardown: runs when the session-scoped loop is closing
        new_val = await agen.__anext__()
        assert new_val == 3
        await agen.aclose()


    async def test_first(async_gen_fixture):
        assert async_gen_fixture == 2


    async def test_second(async_gen_fixture):
        assert async_gen_fixture == 2

How it works under the hood
+++++++++++++++++++++++++++

Understanding the interaction between ``event_loop`` and ``entrypoint``
helps explain why this is safe:

1. When ``event_loop`` is overridden with a wider scope, the loop is
   **not** created by the ``entrypoint`` — it is passed in via the
   ``loop=`` parameter.

2. The ``Entrypoint`` tracks whether it created the loop with an internal
   ``_loop_owner`` flag. When the loop is passed from outside,
   ``_loop_owner`` is ``False``.

3. On teardown, ``Entrypoint.graceful_shutdown()`` only calls
   ``loop.shutdown_asyncgens()`` when ``_loop_owner`` is ``True``.
   This means the function-scoped entrypoint teardown will **not**
   destroy async generators that belong to the wider-scoped loop.

4. ``loop.shutdown_asyncgens()`` is eventually called by the
   ``event_loop`` fixture itself during its own teardown — at the
   correct time, after all fixtures with that scope have been finalized.

Mixing scopes
+++++++++++++

You can have a session-scoped loop with both session-scoped and
function-scoped fixtures. Function-scoped async fixtures will run on
the session loop and be created/destroyed per test as usual:

.. code-block:: python

    from typing import AsyncIterator


    @pytest.fixture(scope="session")
    async def db_pool() -> AsyncIterator[Pool]:
        """Created once, shared across all tests."""
        pool = await create_pool()
        yield pool
        await pool.close()


    @pytest.fixture
    async def db_connection(db_pool: Pool) -> AsyncIterator[Connection]:
        """Created fresh for each test, returned to pool after."""
        async with db_pool.acquire() as conn:
            yield conn


    async def test_query(db_connection: Connection) -> None:
        await db_connection.execute("SELECT 1")

TCPProxy
--------

``TCPProxy`` is a helper for simulating network problems in tests.
It sits between the client and the server, allowing you to add latency,
disconnect clients, or modify traffic on the fly.

.. code-block:: python

    import asyncio
    from typing import AsyncIterator, Iterable
    import pytest
    import aiomisc


    class EchoServer(aiomisc.service.TCPServer):
        async def handle_client(
            self,
            reader: asyncio.StreamReader,
            writer: asyncio.StreamWriter,
        ):
            chunk = await reader.read(65534)
            while chunk:
                writer.write(chunk)
                chunk = await reader.read(65534)
            writer.close()
            await writer.wait_closed()


    @pytest.fixture()
    def server_port(aiomisc_unused_port_factory) -> int:
        return aiomisc_unused_port_factory()


    @pytest.fixture()
    def services(server_port, localhost) -> Iterable[aiomisc.Service]:
        return [EchoServer(port=server_port, address=localhost)]


    @pytest.fixture()
    async def proxy(tcp_proxy, localhost, server_port) -> AsyncIterator["aiomisc.pytest.TCPProxy"]:
        async with tcp_proxy(localhost, server_port) as proxy:
            yield proxy


    async def test_echo(proxy):
        reader, writer = await proxy.create_client()
        writer.write(b"Hello world")
        response = await asyncio.wait_for(reader.read(1024), timeout=1)
        assert response == b"Hello world"


    async def test_disconnect(proxy):
        reader, writer = await proxy.create_client()
        writer.write(b"Hello world")
        await asyncio.wait_for(reader.read(1024), timeout=1)

        await proxy.disconnect_all()
        assert await asyncio.wait_for(reader.read(), timeout=1) == b""


    async def test_slowdown(proxy):
        with proxy.slowdown(read_delay=0.1, write_delay=0.2):
            reader, writer = await proxy.create_client()
            writer.write(b"Hello world")
            response = await asyncio.wait_for(
                reader.read(1024), timeout=2,
            )
            assert response == b"Hello world"


    async def test_content_processor(proxy):
        proxy.set_content_processors(
            lambda _: b"replaced",       # client -> server
            lambda chunk: chunk[::-1],   # server -> client
        )

        reader, writer = await proxy.create_client()
        writer.write(b"original")

        response = await reader.read(16)
        assert response == b"replaced"[::-1]


---

Plugins
=======

aiomisc can be extended with plugins as separate packages. Plugins can
enhance aiomisc by mean of signals_.

.. _signals: #signal

To make your plugin discoverable by aiomisc you should add
``aiomisc.plugins`` entry to ``entry_points`` argument of ``setup``
call in ``setup.py`` of a plugin.

.. code-block:: python

    # setup.py

    setup(
        # ...
        entry_points={
            "aiomisc": ["myplugin = aiomisc_myplugin.plugin"]
        },
        # ...
    )

If you use `pyproject.toml` you might define it like this:

.. code-block:: ini

    [tool.poetry.plugins.aiomisc]
    myplugin = "aiomisc_myplugin.plugin"


Modules provided in ``entry_points`` should have ``setup`` function.
These functions would be called by aiomisc and must contain signals connecting.

If the services are started dynamically, the attached functions will run every
time the services are started and stopped, however, only services that are
currently starting or stopping will be in the ``services`` parameter.

.. code-block:: python
    :name: test_plugin

    # Content of: ``aiomisc_myplugin/plugin.py``
    from typing import Tuple
    from threading import Event

    import aiomisc


    event = Event()
    # Will be shown in ``python -m aiomisc.plugins``
    __doc__ = "Example plugin"


    async def hello(
        *,
        entrypoint: aiomisc.Entrypoint,
        services: Tuple[aiomisc.Service, ...]
    ) -> None:
        print('Hello from aiomisc plugin')
        event.set()


    def setup() -> None:
        """
        This code will be called by loading plugins declared in
        ``pyproject.toml`` or ``setup.py``.
        """
        aiomisc.Entrypoint.PRE_START.connect(hello)

    # Content of: ``my_plugin_example.py``
    # ======================================================================
    # The code below is not related to the plugin, but serves to demonstrate
    # how it works.
    # ======================================================================
    def main():
        """ some function in user code """

        # This function will be called by aiomisc.plugin module
        # in this example it's just for demonstration.
        setup()

        assert not event.is_set()

        with aiomisc.entrypoint() as loop:
            pass

        assert event.is_set()

    main()

    # remove the plugin on when unneeded
    aiomisc.entrypoint.PRE_START.disconnect(hello)


The following signals are available in total:

* ``Entrypoint.PRE_START`` - Will be called before `starting` services.
* ``Entrypoint.PRE_STOP`` - Will be called before `stopping` services.
* ``Entrypoint.POST_START`` - Will be called after services has been `started`.
* ``Entrypoint.POST_STOP`` - Will be called after services has been `stopped`.


List available plugins
----------------------

To see a list of all available plugins, you can call from the
command line ``python -m aiomisc.plugins``:

.. code-block::

    $ python -m aiomisc.plugins
    [11:14:42] INFO     Available 1 plugins.
               INFO     'systemd_watchdog' - Adds SystemD watchdog support to the entrypoint.
    systemd_watchdog

You can also change the behavior and output of the list of modules.
To do this, there are the following flags:

.. code-block::

    $ python3 -m aiomisc.plugins -h
    usage: python3 -m aiomisc.plugins [-h] [-q] [-n]
                                      [-l {critical,error,warning,info,debug,notset}]
                                      [-F {stream,color,json,syslog,plain,journald,rich,rich_tb}]

    optional arguments:
      -h, --help            show this help message and exit
      -q, -s, --quiet, --silent
                            Disable logs and just output plugin-list, alias for
                            --log-level=critical
      -n, --no-output       Disable output plugin-list to the stdout
      -l {critical,error,warning,info,debug,notset}, --log-level {critical,error,warning,info,debug,notset}
                            Logging level
      -F {stream,color,json,syslog,plain,journald,rich,rich_tb}, --log-format {stream,color,json,syslog,plain,journald,rich,rich_tb}
                            Logging format

Here are some run examples.

.. code-block::

    $ python3 -m aiomisc.plugins -n
    [12:25:57] INFO     Available 1 plugins.
               INFO     'systemd_watchdog' - Adds SystemD watchdog support to the entrypoint.

This prints human-readable list of plugins and its descriptions.


.. code-block::

    $ python3 -m aiomisc.plugins -s
    systemd_watchdog

This useful for ``grep`` or other pipelining tools.

The default prints both, the human-readable log to stderr and the list
of plugins to stdout, so you can use this without options in a pipeline,
and read the list to stderr.