Skip to main content

Reducer for similar simultaneously coroutines

Project description

Python versions PyPI version coverage report

About Async-Reduce

async_reduce(coroutine) allows aggregate all similar simultaneous ready to run coroutines and reduce to running only one coroutine. Other aggregated coroutines will get result from single coroutine.

It can boost application performance in highly competitive execution of the similar asynchronous operations and reduce load for inner systems.

Quick example

from async_reduce import async_reduce


async def fetch_user_data(user_id: int) -> dict:
    """" Get user data from inner service """
    url = 'http://inner-service/user/{}'.format(user_id)

    return await http.get(url, timeout=10).json()


@web_server.router('/users/(\d+)')
async def handler_user_detail(request, user_id: int):
    """ Handler for get detail information about user """

    # all simultaneous requests of fetching user data for `user_id` will 
    # reduced to single request
    user_data = await async_reduce(
        fetch_user_data(user_id)
    )

    # sometimes ``async_reduce`` cannot detect similar coroutines and
    # you should provide special argument `ident` for manually determination
    user_statistics = await async_reduce(
        DataBase.query('user_statistics').where(id=user_id).fetch_one(),
        ident='db_user_statistics:{}'.format(user_id)
    )

    return Response(...)

In that example without using async_reduce if client performs N simultaneous requests like GET http://web_server/users/42 web_server performs N requests to inner-service and N queries to database. In total: N simultaneous requests emits 2 * N requests to inner systems.

With async_reduce if client performs N simultaneous requests web_server performs one request to inner-service and one query to database. In total: N simultaneous requests emit only 2 requests to inner systems.

See other real examples.

Similar coroutines determination

async_reduce(coroutine) tries to detect similar coroutines by hashing local variables bounded on call. It does not work correctly if:

  • one of the arguments is not hashable
  • coroutine function is a method of class with specific state (like ORM)
  • coroutine function has closure to unhashable variable

You can disable auto-determination by setting custom key to argument ident.

Use as decorator

Also library provide special decorator @async_reduceable(), example:

from async_reduce import async_reduceable


@async_reduceable()
async def fetch_user_data(user_id: int) -> dict:
    """" Get user data from inner service """
    url = 'http://inner-servicce/user/{}'.format(user_id)

    return await http.get(url, timeout=10).json()


@web_server.router('/users/(\d+)')
async def handler_user_detail(request, user_id: int):
    """ Handler for get detail information about user """
    return await fetch_user_data(user_id)

Hooks

Library supports hooks. Add-on hooks:

  • DebugHooks - print about all triggered hooks
  • StatisticsOverallHooks - general statistics on the use of async_reduce
  • StatisticsDetailHooks - like StatisticsOverallHooks but detail statistics about all coroutine processed by async_reduce

Example:

from async_reduce import AsyncReducer
from async_reduce.hooks import DebugHooks

# define custom async_reduce with hooks
async_reduce = AsyncReducer(hooks=DebugHooks())


async def handler_user_detail(request, user_id: int):
    user_data = await async_reduce(fetch_user_data(user_id))

See more detail example in examples/example_hooks.py.

You can write custom hooks via inherit from BaseHooks.

Caveats

  • If single coroutine raises exceptions all aggregated coroutines will get same exception too

  • If single coroutine is stuck all aggregated coroutines will stuck too. Limit execution time for coroutine and add retries (optional) to avoid it.

  • Be careful when return mutable value from coroutine because single value will shared. Prefer to use non-mutable value as coroutine return.

Development

See DEVELOPMENT.md.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for async-reduce, version 0.2.1
Filename, size File type Python version Upload date Hashes
Filename, size async_reduce-0.2.1-py3-none-any.whl (12.1 kB) File type Wheel Python version py3 Upload date Hashes View hashes
Filename, size async_reduce-0.2.1.tar.gz (9.1 kB) File type Source Python version None Upload date Hashes View hashes

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page