Skip to main content

A simple throttling package

Project description

This library is inspired by this book and this implementation


  1. Set capacity(max parallel requests) and queue(max queued requests) limits.
  2. Per-consumer limits. For instance, to not allow any consumer to use more than 70% of service's capacity.
  3. Per-request priorities. For instance, to not allow requests with lowest priority to be queued or to now allow requests with normal priority to use more than 90% of service's capacity.


from aio_throttle import Throttler, MaxFractionCapacityQuota, ThrottlePriority, ThrottleResult

capacity_limit = 100
queue_limit = 200
consumer_quotas = [MaxFractionCapacityQuota(0.7)]
priority_quotas = [MaxFractionCapacityQuota(0.9, ThrottlePriority.NORMAL)]
throttler = Throttler(capacity_limit, queue_limit, consumer_quotas, priority_quotas)

async with throttler.throttle(consumer, priority) as result:
    ... # check if result is ThrottleResult.ACCEPTED or not

Example of integration with aiohttp:

from typing import Awaitable, Callable

from aiohttp import web
from aiohttp.web_middlewares import middleware
from aiohttp.web_request import Request
from aiohttp.web_response import StreamResponse

from aio_throttle import Throttler, ThrottlePriority

Handler = Callable[[Request], Awaitable[StreamResponse]]

async def middleware(request: Request, handler: Handler) -> StreamResponse:
    throttler: Throttler =['throttler']
    service = request.headers.get('X-Service', 'UNKNOWN')
    priority = ThrottlePriority(request.headers.get('X-Request-Priority', ThrottlePriority.NORMAL))
    async with throttler.throttle(service, priority) as result:
        if not result:
            return web.Response(status=503)
        return await handler(request)

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 aio-throttle, version 1.2.1
Filename, size File type Python version Upload date Hashes
Filename, size aio_throttle-1.2.1-py3-none-any.whl (6.1 kB) File type Wheel Python version py3 Upload date Hashes View hashes
Filename, size aio-throttle-1.2.1.tar.gz (4.2 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 DigiCert DigiCert EV certificate StatusPage StatusPage Status page