async-services 1.0.2

Synchronus Wrapper for Async Code

Async Services

Run fast asynchronous code from a synchronous code. Async Services provide a synchronous wrapper to run third party asynchronous code or any coroutine for that matter in a synchronous way from a synchronous code.

Installation

pip install async_services

For Development Purposes

Install project Dependencies

pip install -r requirements.txt

pip install -U .

Running the tests

You can run the tests with the following command

pytest .

And coding style tests

pycodestyle .

Example Usage

from async_services.core import run_coro, run_manager, stop_manager
from async_services.core.manager import CoroStatus
import asyncio

async def coroutine(seconds=1, raise_exception=False):
    await asyncio.sleep(seconds)
    if raise_exception:
        raise Exception("Sample Exception")
    return "Hello World"

run_manager()
result = run_coro(coroutine(), block=True)
print(result)
assert result[0] == CoroStatus.Completed
assert result[1] == "Hello World"
stop_manager()

Output

Result will be a tuple consisting of two values (status, result) status will be a integer between 0 and 5 and it defines the state of the coruotine

(1, 'Hello World')

Coroutine Status

Presently there are six status of a coruotine

1. Queued = 0 -> Coroutine is still queued waiting to be executed or is being executed
2. Completed = 1 -> Coroutine has Completed Successfully
3. Failed = 2 -> Coroutine Completed Successfully , But callback function raised an exception
4. Cancelled = 3 -> Coroutine was Cancelled
5. Timeout = 4 -> Coroutine did not complete in the given time
6. CoroutineException = 5 -> Coroutine Itself Raised an Exception

Authors

• Ankit Kathuria - Initial work

License

This project is licensed under the MIT License - see the LICENSE.md file for details

Acknowledgments

• Hat tip to anyone whose code was used
• Inspiration
• etc