Reference

Application Start

ponty.startmeup(*, port, providers=(), route_tables=())

The main entrypoint for your application. Kicks off the server event loop.

Parameters

port (int) – TCP/IP port for HTTP server. No default, must be provided
providers (Sequence[Provider]) – see Providers
route_tables (Sequence[RouteTableDef]) – Optional sequence of aiohttp.web.RouteTableDef instances

Return type

None

Routes

Ponty relies on decorator-style routing. Simply wrap your handlers like so:

from ponty import get, render_json


@get("/hello")
@render_json
async def handler(_):
    return {"greeting": "hello world"}

Ponty makes no effort to intercept the aiohttp.web.Request instance provided by aiohttp. It does, however, provide a Request class of its own to simplify processing.

Ponty handlers may return aiohttp.web.Response instances directly, or they may take advantage of utilities such as ponty.render_json() or ponty.raise_status().

@ponty.route(method, path, **kw)

Bind a (method, path) pair to its handler, and register with the master route table.

PontyError’s are automatically trapped and handled here.

Parameters

method (str) – HTTP method, e.g. GET
path (str) – Relative path to resource. May contain braces {} for dynamic components, or references to RouteParameter instances in an f-string
kw – Additional arguments discussed here

@ponty.get(path, **kw): Register a GET handler. See route() for parameter information.

@ponty.post(path, **kw): Register a POST handler. See route() for parameter information.

@ponty.put(path, **kw): Register a PUT handler. See route() for parameter information.

@ponty.delete(path, **kw): Register a DELETE handler. See route() for parameter information.

@ponty.patch(path, **kw): Register a PATCH handler. See route() for parameter information.

ponty.route_iter()

Iterator, over routes on the primary route table.

Returns: (method, path) pairs
Return type: Iterator[tuple[str, str]]

Request

class ponty.Request(req)

Base class for request pre-processing. Instantiated by expect().

Request Fields for extracting components of the request should be stored as class variables on subclasses.

Parameters: req (aiohttp.web.Request) – automatically supplied by expect()

@ponty.expect(cls, *, mimetype=None)

Preprocess the HTTP request, according to the rules configured in the Request subclass.

Parameters

cls (type[Request]) – subclass of Request, with rules for processing the HTTP request attached as descriptors
mimetype (str) – expected IANA media type. If provided and the specified type does not match the media type found in the content-type header, throws a 415

For example,

 from ponty import (
     expect,
     get,
     render_json,
     Request,
     StringRouteParameter,
 )


 class HelloReq(Request):

     name = StringRouteParameter()


 @get(f"/hello/{HelloReq.name}")  # evaluates to "/hello/{name:\w+}"
 @expect(HelloReq)
 @render_json
 async def greet(name: str):
     return {"greeting": f"hello, {name}!"}

 $ curl localhost:8080/hello/you -v | python -m json.tool
 > GET /hello/you HTTP/1.1
 > Host: localhost:8080
 > User-Agent: curl/7.64.1
 > Accept: */*
 >
 < HTTP/1.1 200 OK
 < Content-Type: application/json; charset=utf-8
 < Content-Length: 72
 < Date: Thu, 11 Aug 2022 02:58:44 GMT
 < Server: Python/3.9 aiohttp/3.7.3
 <
 {
     "data": {
         "greeting": "hello, you!"
     },
     "elapsed": 0,
     "now": 1660186724077
 }

Request Fields

So far we’ve looked at the overall request. Now let’s look at the individual fields on that request.

Ponty uses Python descriptors for extracting and processing components of the HTTP request. In conjunction with the ponty.expect() decorator, request fields are parameterized directly into the decorated handler. (Descriptor names are reused as parameter names.)

For a more detailed understanding of descriptors, see the Python docs.

Route Parameters

class ponty.RouteParameter(*, pattern, cast_func=None)

The RouteParameter descriptor is used to identify and extract variable resources from the URI. It is generic on one variable.

When invoked from an instance, returns the variable part of the resolved route.

When invoked from a class, returns the {identifier:regex} rendering required for path matching. (See Variable Resources.) Particularly useful in an f-string (example below).

Parameters

pattern (str) – custom regular expression to match the variable part
cast_func (Optional[Callable[[str], _T]]) – converts the variable part to type T

New reusable “matchers” can be created like so:

 from ponty import (
     expect,
     get,
     Request,
     RouteParameter,
 )


 class FiveLetters(RouteParameter[str]):

     def __init__(self):
         super().__init__(pattern="[a-zA-Z]{5}")


 class MyReq(Request):

     id = FiveLetters()


 @get(f"/obj/{MyReq.id}")
 @expect(MyReq)
 async def handler(id: str):
     ...

class ponty.PosIntRouteParameter: Inherits RouteParameter. Matches on a sequence of digits, \d+.

class ponty.StringRouteParameter: Inherits RouteParameter. Matches on strings, \w+.

Query Parameters

class ponty.QueryParameter(*, key: str = '', required: Literal[False] = False, default: _T, cast_func: Callable[[str], _T], values: Iterable[str] = ())

class ponty.QueryParameter(*, key: str = '', required: Literal[True], cast_func: Callable[[str], _T], values: Iterable[str] = ())

Extracts the value of query parameter with name key from the URI, if it’s provided. Generic on one variable.

Parameters

key (str) – if provided, query parameter key name. By default, uses descriptor __set_name__ to fetch the variable name
required (bool) – if True, the query parameter will be treated as a mandatory component of the request; an HTTP 400 will be raised in the event no value is supplied. Default False
default (_T) – default value, returned if the query param has not been provided. Set required=True instead if this should be treated as an error condition. One of required or default must be provided
cast_func (Callable[[str], _T]) – function to convert the string value provided to type T. If the function raises a ValueError when it fails, the error will be trapped and reraised as an HTTP 400
values (Iterable[str]) – if supplied, validates the captured query parameter against these legal values. Throws a 400 in the event of a mismatch

It’s easiest to use StringQueryParameter or PosIntQueryParameter for simple cases in practice.

Use the QueryParameter base class to create new custom parsers, in the same way as RouteParameter:

from ponty import (
    expect,
    get,
    render_json,
    QueryParameter,
    Request,
)


_boolish_vals: dict[str, bool] = {
    "1": True,
    "0": False,
    "yes": True,
    "no": False,
    "true": True,
    "false": False,
    "t": True,
    "f": False,
}


class BoolQueryParam(QueryParameter[bool]):

    def __init__(self, **kw):
        super().__init__(
            cast_func=_boolish_vals.__getitem__,
            values=_boolish_vals.keys(),
            **kw
        )


class HelloReq(Request):

    capitalize = BoolQueryParam(default=False)


@get("/hello")
@expect(HelloReq)
@render_json
async def greet(capitalize: bool):
    greeting = "hello world"
    if capitalize:
        greeting = greeting.upper()
    return {"greeting": greeting}

the default

$ curl localhost:8080/hello | python -m json.tool
{
    "data": {
        "greeting": "hello world"
    },
    "elapsed": 0,
    "now": 1679444906510
}

providing a non-default option

$ curl 'localhost:8080/hello?capitalize=yes' | python -m json.tool
{
    "data": {
        "greeting": "HELLO WORLD"
    },
    "elapsed": 0,
    "now": 1679444906739
}

providing an illegal value

$ curl 'localhost:8080/hello?capitalize=blah' -v
> GET /hello?capitalize=blah HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.79.1
> Accept: */*
>
< HTTP/1.1 400 Bad Request
< Content-Type: text/plain; charset=utf-8
< Content-Length: 46
< Date: Wed, 22 Mar 2023 00:21:44 GMT
< Server: Python/3.9 aiohttp/3.7.3
<
capitalize must be one of {1,0,yes,no,true,false,t,f}

class ponty.StringQueryParameter(**kw)

Inherits QueryParameter. Treats the captured query param as a string.

from ponty import (
    expect,
    get,
    render_json,
    Request,
    StringQueryParameter,
)


class HelloReq(Request):

    punc = StringQueryParameter(default="!")


@get("/hello")
@expect(HelloReq)
@render_json
async def greet(punc: str):
    return {"greeting": f"hello world{punc}"}

the default in action

$ curl localhost:8080/hello | python -m json.tool
{
    "data": {
        "greeting": "hello world!"
    },
    "elapsed": 0,
    "now": 1660439305592
}

overriding the default

$ curl 'localhost:8080/hello?punc=.' | python -m json.tool
{
    "data": {
        "greeting": "hello world."
    },
    "elapsed": 0,
    "now": 1660439305592
}

A similar example, using values:

from ponty import (
    expect,
    get,
    render_json,
    Request,
    StringQueryParameter,
    StringRouteParameter,
)


_hellos: dict[str, str] = {
    "en": "hello",
    "es": "hola",
    "no": "hallo",
}


class HelloReq(Request):

    name = StringRouteParameter()
    lang = StringQueryParameter(
        values=_hellos.keys(),
        default="en",
    )


@get(f"/hello/{HelloReq.name}")
@expect(HelloReq)
@render_json
async def ahoy(name: str, lang: str):
    # no need to trap the KeyError here, the descriptor has already
    # validated `lang` is a key in `_hellos`
    hi = _hellos[lang]
    return {"greeting": f"{hi} {name}!"}

$ curl 'localhost:8080/hello/muchacho?lang=es' | python -m json.tool
{
    "data": {
        "greeting": "hola muchacho!"
    },
    "elapsed": 0,
    "now": 1660440618107
}

class ponty.PosIntQueryParameter(**kw)

Inherits QueryParameter. Casts the captured query param to an integer, and validates it is non-negative.

from ponty import (
    expect,
    get,
    render_json,
    Request,
    PosIntQueryParameter,
)


class HelloReq(Request):

    num_exclamations = PosIntQueryParameter(key="bangs", default=1)


@get("/hello")
@expect(HelloReq)
@render_json
async def greet(num_exclamations: int):
    return {"greeting": f"hello world{'!' * num_exclamations}"}

$ curl localhost:8080/hello?bangs=7 | python -m json.tool
{
    "data": {
        "greeting": "hello world!!!!!!!"
    },
    "elapsed": 0
    "now": 1681345852404,
}

not an int

$ curl localhost:8080/hello?bangs=abc
> GET /hello?bangs=abc HTTP/1.1
> Host: localhost:8008
> User-Agent: curl/7.79.1
> Accept: */*
>
< HTTP/1.1 400 Bad Request
< Content-Type: text/plain; charset=utf-8
< Content-Length: 21
< Date: Thu, 13 Apr 2023 00:31:40 GMT
< Server: Python/3.9 aiohttp/3.7.3
<
'abc' could not be cast

Request Body

class ponty.TextBody: Contains the raw request body.

class ponty.JsonBody: Contains the deserialized request body.

class ponty.ValidatedJsonBody(*, filepath=None)

Validates the deserialized request body against the given jsonschema. Invalid input immediately triggers a 400 response.

Inherits JsonBody.

Parameters

filepath (Optional[str]) – jsonschema file location (absolute, relative to PYTHONPATH, or relative to working directory).
schema (Optional[dict[str, Any]]) –

validator.json

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "properties": {
    "first_name": { "type": "string" },
    "last_name": { "type": "string" }
  },
  "required": ["first_name", "last_name"]
}

from ponty import (
    expect,
    post,
    render_json,
    Request,
    ValidatedJsonBody,
)


class MyReq(Request):

    body = ValidatedJsonBody(filepath="validator.json")


@post("/test")
@expect(MyReq)
@render_json
async def handler(body):
    return {"name": f"{body['last_name']}, {body['first_name']}"}

success

$ curl localhost:8080/test \
    -H "content-type:application/json" \
    -d '{"first_name": "Donald", "last_name": "Duck"}' | python -m json.tool
{
    "data": {
        "name": "Duck, Donald",
    },
    "elapsed": 0,
    "now": 1660505985510
}

error, “last_name” omitted

$ curl localhost:8080/test \
    -H "content-type:application/json" \
    -d '{"first_name": "Donald"}' \
    -v
...
< HTTP/1.1 400 Bad Request
< Content-Type: text/plain; charset=utf-8
...
'last_name' is a required property

class ponty.ParsedJsonBody(cls, *, filepath=None)

Holds the validated request body, structured as an instance of cls.

Inherits ValidatedJsonBody.

Parameters

cls (dataclass) – dataclass, into which the request body is marshalled. A custom jsonschema validator is automatically created by ponty.dataclass_to_jsonschema()
filepath (Optional[str]) – if provided, path to the jsonschema file (see ValidatedJsonBody). Supplants the dataclass-built default

from dataclasses import asdict, dataclass
import typing

from ponty import (
    expect,
    post,
    render_json,
    Annotation,
    ParsedJsonBody,
    Request,
)


@dataclass(frozen=True)
class Person:

    first_name: str
    last_name: str
    favorite_color: typing.Optional[str]
    height: typing.Annotated[int, Annotation(description="cm")]

    class JsonSchema:
        annotation = Annotation(description="request shape for a person")


class NewPersonReq(Request):

    person = ParsedJsonBody(Person)


@post("/person")
@expect(NewPersonReq, mimetype="application/json")
@render_json
async def handler(person: Person):
    return asdict(person)

the generated jsonschema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "properties": {
    "first_name": {
      "type": "string"
    },
    "last_name": {
      "type": "string"
    },
    "favorite_color": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ]
    },
    "height": {
      "type": "integer",
      "description": "cm"
    }
  },
  "title": "Person",
  "type": "object",
  "description": "request shape for a person",
  "required": [
    "first_name",
    "last_name",
    "favorite_color",
    "height"
  ]
}

(note the highlighted lines, controlled by Annotation)

success, echo back the request body

$ curl localhost:8080/person \
    -H "content-type:application/json" \
    -d '{"first_name": "Mickey", "last_name": "Mouse", "favorite_color": "blue", "height": 1000}' \
    -v | python -m json.tool
> POST /person HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.64.1
> Accept: */*
> content-type:application/json
> Content-Length: 83
>
< HTTP/1.1 200 OK
< Content-Type: application/json; charset=utf-8
< Content-Length: 129
< Date: Sun, 14 Aug 2022 21:04:35 GMT
< Server: Python/3.9 aiohttp/3.7.3
<
{
    "data": {
        "favorite_color": "blue",
        "first_name": "Mickey",
        "height": 1000,
        "last_name": "Mouse"
    },
    "elapsed": 0,
    "now": 1660511075634
}

error, missing “height”

$ curl localhost:8080/person \
    -H "content-type:application/json" \
    -d '{"first_name": "Mickey", "last_name": "Mouse", "favorite_color": "blue"}' \
    -v
...
< HTTP/1.1 400 Bad Request
< Content-Type: text/plain; charset=utf-8
...
'height' is a required property

error, wrong type for “height”

$ curl localhost:8080/person \
    -H "content-type:application/json" \
    -d '{"first_name": "Mickey", "last_name": "Mouse", "favorite_color": "blue", "height": "abc"}' \
    -v
...
< HTTP/1.1 400 Bad Request
...
'abc' is not of type 'integer'

Request Headers

class ponty.Header(*, key, required=False, default='')

Extracts request header key.

Parameters

key (str) – header name
required (bool) – if True, throws a 400 if the request header is not provided
default (str) – default value, if the header does not appear on the request

class ponty.ContentLength(**kw): Inherits Header. Extracts the “content-length” header.

class ponty.ContentType(**kw): Inherits Header. Extracts the “content-type” header.

Other

class ponty.AIOHttpReq: Forwards along the aiohttp.web.Request.

class ponty.Cookie(*, name, required=False, default='', errorcode=400)

Extracts request cookie name.

Parameters

name (str) – cookie name
required (bool) – if True, throws a 400 if the cookie is not provided
default (str) – default value, if the cookie is not present
errorcode (int) – HTTP status code raised, if the cookie is required and not supplied

Schema Validation

ponty.dataclass_to_jsonschema(cls)

Generate the jsonschema for a dataclass.

Optionally, fields can be enriched by passing jsonschema keywords to Annotation.

Additionally, the dataclass can be enriched with keywords by embedding a JsonSchema class (using Annotation as well).

Parameters: cls (dataclass) – dataclass definition
Raises: ValueError – raised if cls is not a dataclass
Returns: the JSON schema
Return type: dict[str, Any]

from dataclasses import dataclass
import json
import typing

from ponty import Annotation, dataclass_to_jsonschema


@dataclass
class Model:

    name: str
    body_style: typing.Literal["coupe", "sedan", "hatchback", "convertible", "wagon", "suv"]
    year: typing.Annotated[int, Annotation(minimum=1886, maximum=2030)]
    msrp: typing.Annotated[int, Annotation(description="MSRP, dollars.")]
    cylinders: int = 4


@dataclass
class Make:

    name: typing.Annotated[str, Annotation(description="Brand name")]
    models: list[Model]

    class JsonSchema:
        annotation = Annotation(description="Automotive brand", examples=["Toyota", "Honda"])


if __name__ == "__main__":
    schema = dataclass_to_jsonschema(Make)
    print(json.dumps(schema, indent=2))

generated jsonschema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "properties": {
    "name": {
      "type": "string",
      "description": "Brand name"
    },
    "models": {
      "type": "array",
      "items": {
        "allOf": [
          {
            "$ref": "#/$defs/Model"
          }
        ]
      }
    }
  },
  "title": "Make",
  "type": "object",
  "description": "Automotive brand",
  "examples": [
    "Toyota",
    "Honda"
  ],
  "required": [
    "name",
    "models"
  ],
  "$defs": {
    "Model": {
      "properties": {
        "name": {
          "type": "string"
        },
        "body_style": {
          "enum": [
            "coupe",
            "sedan",
            "hatchback",
            "convertible",
            "wagon",
            "suv"
          ]
        },
        "year": {
          "type": "integer",
          "minimum": 1886,
          "maximum": 2030
        },
        "msrp": {
          "type": "integer",
          "description": "MSRP, dollars."
        },
        "cylinders": {
          "type": "integer",
          "default": 4
        }
      },
      "title": "Model",
      "type": "object",
      "required": [
        "name",
        "body_style",
        "year",
        "msrp"
      ]
    }
  }
}

class ponty.Annotation: Enrich member variable annotations with context-specific metadata. These annotations appear in the jsonschema as generic keywords; in some cases they are used for additional validation (e.g. array.minItems), while in others they simply help make the schema self-documenting (e.g. description).

Response

@ponty.render_json

Encode the decorated function’s return value with json.dumps, and package it into an HTTP response. The JSON has the form:

{
    "data": <function return value>,
    "elapsed": <time elapsed, in millis>,
    "now": <current epoch timestamp, in millis>,
}

e.g.,

from ponty import get, render_json


@get("/hello")
@render_json
async def handler(_):
    return {"greeting": "hello world!"}

$ curl localhost:8080/hello | python -m json.tool
{
    "data": {
        "greeting": "hello world!"
    },
    "elapsed": 1,
    "now": 1660440618107
}

ponty.raise_status(status, *, text=None, body=None, content_type=None, **kw)

Raise an aiohttp.web.HTTPException for the given status code.

Parameters

status (int) – HTTP status code
text (Optional[str]) – response body
body (Optional[Any]) – json-serializable response body. Automatically sets the content-type header to ‘application/json’. Only one of text or body may be supplied.
content_type (Optional[str]) – value for the response’s content-type header
kw –
additional params for aiohttp.web.HTTPException

Return type

NoReturn

@get("/fail")
async def handler(_):
    raise_status(400, body={"a": 1})

$ curl localhost:8080/fail -v
> GET /fail HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.64.1
> Accept: */*
>
< HTTP/1.1 400 Bad Request
< Content-Type: application/json; charset=utf-8
< Content-Length: 8
< Date: Wed, 10 Aug 2022 02:28:55 GMT
< Server: Python/3.9 aiohttp/3.7.3
<
{"a": 1}

HTTP Errors

exception ponty.PontyError(*, text=None, body=None, **kw)

PontyError’s may be raised anywhere within your program to immediately respond with an HTTP status code. Unhandled errors are reraised as aiohttp.web.HTTPException.

Parameters

text (Optional[str]) – response body, as text/plain
body (Any) – json-serializable response body
kw – see parameter description for aiohttp.web.HTTPException

status_code: int = 500: HTTP status code for the response. See the official IANA registry here

exception ponty.DoesNotExist(*, text=None, body=None, **kw)

Returns a 404. Inherits PontyError.

Parameters

text (Optional[str]) –
body (Any) –

exception ponty.ValidationError(*, text=None, body=None, **kw)

Returns a 400. Inherits PontyError.

Parameters

text (Optional[str]) –
body (Any) –

Providers

Providers manage shared assets, and are Ponty’s way of guaranteeing assets are cleaned up upon program exit 1.

Providers are asynchronous generators that take a single argument, an instance of aiohttp.web.Application. Code before the yield initializes the asset during startup, while code after the yield runs during teardown.

Providers must be passed to ponty.startmeup() for proper handling.

ponty.http_client_provider(*, name='_default', timeout=5, concurrency=100, headers=None, cookies=None, **kw)

Builds an HTTP Client Session provider.

Wraps aiohttp.ClientSession.

Parameters

name (str) – unique name, used by lease_http_client()
timeout (int) – total timeout for the complete operation, in seconds
concurrency (int) – maximum number of concurrent requests
headers (Optional[dict[str, Any]]) – headers to send with each request
cookies (Optional[dict[str, Any]]) – cookies to send with each request
kw – additional parameters, specified here

Returns

a provider, which should be passed to startmeup()

Return type

Callable[[Application], AsyncIterator[None]]

startmeup(
    port=8080,
    providers=(
        http_client_provider(
            timeout=30,
            concurrency=10,
        ),
    ),
)

ponty.lease_http_client(name='_default')

Context manager. Fetch the named http client session, leasing one of its request positions or blocking until one becomes available, then return it to the pool after the block is exited.

Parameters: name (str) – see http_client_provider() parameter name
Raises: KeyError – if the named session was not registered (via startmeup())
Yields: aiohttp.ClientSession
Return type: AsyncIterator[ClientSession]

async with lease_http_client() as sess:
    async with sess.get("https://www.python.org/") as resp:
        html = await resp.text()

Utils

@ponty.retry(*excs, max_retries=1, t1_ms=100, t2_ms=4000, backoff_factor=1, anticollision=False)

Retry the wrapped function if the specified exception(s) are raised.

Supports sync & async functions.

Parameters

excs (type[Exception]) – Exception class(es) to trap
max_retries (int) – max number of retries to attempt
t1_ms (int) – initial interval between retries, in millis
t2_ms (int) – max interval between retries, in millis
backoff_factor (float) –
rate reduction factor for exponential backoff \(t = b^c\), where

t is the delay factor between calls

b = backoff_factor is the base

c is the number of failures observed so far

\(b > 1\) will reduce the retry rate, \(b = 1\) will give consistent retry intervals, and \(b < 1\) will accelerate retries (unusual choice). Set to 2 for run-of-the-mill binary exponential backoff.
anticollision (bool) – a deterministic algorithm may be unsuitable when errors are caused by collisions, because each client will sleep for the same amount of time, leading to subsequent collisions ad infinitum. Pass True to treat \(b^c\) as an upper bound on the time delay; in this case we will sleep for \(k * t1\_ms\), where k is a random integer on \([0, \lfloor b^c \rfloor)\).

import random
import time

from ponty import retry


@retry(KeyError, ValueError, max_retries=10, backoff_factor=2)
def getval():
    i = random.choice(range(5))
    print(f"i = {i}, now = {time.time()}")

    if i == 0:
        return 42
    if i % 2:
        raise KeyError
    raise ValueError

>>> print(getval())
i = 3, now = 1660102025.223419
i = 3, now = 1660102025.328576
i = 4, now = 1660102025.533844
i = 1, now = 1660102025.939289
i = 1, now = 1660102026.744708
i = 2, now = 1660102028.347733
i = 0, now = 1660102031.549533
42

Memoization

Ponty caches are extensible, async-friendly memoizers that help to limit stampedes. They are function decorators, like @functools.cache and @functools.lru_cache, and may be a good fit for cases where the builtins are insufficient (e.g. stampede concerns, the use-case demands a TTL, or perhaps the cache needs to be shared across a fleet of instances behind a load balancer).

In the box

@ponty.memo.localcache(*, ttl_ms=0, maxwait_ms=1000, pulse_ms=50, maxsize=128, name='')

Process-RAM LRU memoizer with antistampede.

Good for small frequently-used datasets (high ttl) or volatile stampede-likely objects (low ttl).

Parameters

ttl_ms (int) – millis to expiry. Use 0 (the default) for no expiry
maxwait_ms (int) – millis to wait for a lock to resolve. Throws Stampede error when \((n * pulse\_ms) > maxwait\_ms\)
pulse_ms (int) – antistampede recheck frequency
maxsize (int) – evict least recently used values once the cache reaches maxsize elements
name (str) – Optional. Providing a name registers the cache, so it can be used by invalidate()

Here is an example that demonstrates stampede control. With maxwait_ms = 0, this operates as a mandatory lock:

import asyncio

from ponty import get, render_json
from ponty.memo import localcache


@get("/cachetest")
@render_json
async def _locktest(_):
    return await _fetch()


@localcache(maxwait_ms=0)
async def _fetch():
    await asyncio.sleep(1)  # some expensive operation
    return {"key": "value"}

For five simultaneous requests, the first acquires the lock and begins work while the others are immediately declined:

$ for i in {1..5}; do curl localhost:8080/cachetest -v & done

< HTTP/1.1 409 Conflict
< Date: Tue, 23 Aug 2022 21:49:12 GMT

< HTTP/1.1 409 Conflict
< Date: Tue, 23 Aug 2022 21:49:12 GMT

< HTTP/1.1 409 Conflict
< Date: Tue, 23 Aug 2022 21:49:12 GMT

< HTTP/1.1 409 Conflict
< Date: Tue, 23 Aug 2022 21:49:12 GMT

< HTTP/1.1 200 OK
< Date: Tue, 23 Aug 2022 21:49:13 GMT
{"now": 1661291352844, "data": {"key": "value"}, "elapsed": 1006}

Now that the item is cached, running this again gives five immediate cache hits:

$ for i in {1..5}; do curl localhost:8080/cachetest & done
{"now": 1661291746863, "data": {"key": "value"}, "elapsed": 0}
{"now": 1661291746863, "data": {"key": "value"}, "elapsed": 0}
{"now": 1661291746863, "data": {"key": "value"}, "elapsed": 0}
{"now": 1661291746863, "data": {"key": "value"}, "elapsed": 0}
{"now": 1661291746863, "data": {"key": "value"}, "elapsed": 0}

Bump to maxwait_ms = 2000 and clear the cache:

$ for i in {1..5}; do curl localhost:8080/cachetest & done
{"now": 1661396864744, "data": {"key": "value"}, "elapsed": 1003}
{"now": 1661396864744, "data": {"key": "value"}, "elapsed": 1043}
{"now": 1661396864744, "data": {"key": "value"}, "elapsed": 1044}
{"now": 1661396864744, "data": {"key": "value"}, "elapsed": 1045}
{"now": 1661396864744, "data": {"key": "value"}, "elapsed": 1045}

The first request is the only one doing work here; the others wait up to two seconds, checking in every pulse_ms = 50 milliseconds (the default) to see if a result is available.

ponty.memo.invalidate(cachename, *a, **kw)

Context manager. Invalidates an item in the cache.

Waits for then holds the cache’s lock until the context manager exits. This prevents simultaneous fetches from re-cache-ing the item before mutations are committed. This also means functions calling ‘invalidate’ CANNOT be wrapped with a cache decorator, as the decorator will already hold the lock.

Parameters: cachename (str) – name of the registered cache. Must match cache()’s name parameter
Return type: AsyncIterator[None]

Note the *arg/**kwarg combo must match the arguments to the cache-decorated function EXACTLY in order to get a key hit. E.g.,

@localcache(name="foo")
async def get_foo(foo_id):
    ...


async def update_foo(foo_id, ...):
    async with invalidate("foo", foo_id):  # this hits
        ...
    async with invalidate("foo", foo_id=foo_id):  # this does not
        ...

Building blocks

Tools for building custom caching components.

@ponty.memo.cache(store, antistampede, name='')

Cache the decorated function’s return value. Not thread-safe.

As a rule of thumb, storage mechanics for a cache are implemented in store, and synchronization rules are handled by antistampede. See the localcache() source ↗ for an example.

Parameters

store (CacheStore) – container, implementing get/set mechanics for cached items
antistampede (Lock) – instance of ponty.memo.Lock; blocks on simultaneous access to the requested key to avoid a stampede
name (str) – if provided, registers the cache for use by invalidate(). Must be unique

class ponty.memo.CacheStore

Abstract Base Class. Container for cached items.

cache() expects two abstract methods, get and set, and invalidate() expects one, remove. All three must return awaitables.

Generic on one variable, the type T of the cached item.

abstract async get(key)

Fetch the cached value.

Errors are not captured.

Parameters: key (str) – unique id for the cached item
Returns: the cached item, or the cachemiss sentinel if it does not exist
Return type: Union[T, type[cachemiss]]

abstract async remove(key)

Remove an item from the cache.

Parameters: key (str) – unique id for the cached item
Returns: True if the item is found and removed, False otherwise
Return type: bool

abstract async set(key, data)

Add an item to the cache.

Parameters

key (str) – unique id for the item
data (_T) – the item to be cached

Return type

None

class ponty.memo.cachemiss

Sentinel value representing a cache miss.

Returned by subclasses of CacheStore.

exception ponty.memo.Stampede(*, text=None, body=None, **kw)

Raised when the cache comes under high load and the lock times out.

Inherits Locked.

Parameters

text (Optional[str]) –
body (Any) –

Locking

Ponty locks are primarily intended for stampede control in cache(), but may also be used independently to enforce access limits on e.g. shared state. Whenever possible, simply use asyncio.Lock.

As a rule of thumb, wrap the minimum amount of code necessary in lock context blocks. This will reduce the frequency and duration of one coroutine blocking another.

In the box

ponty.memo.locallock(maxwait_ms=0, pulse_ms=100, timeout_error=Locked)

Mutex; coordinates across coroutines. Not thread-safe.

Parameters

maxwait_ms – total milliseconds to wait, before raising an exception with the caller. Raises timeout_error when \(n * pulse\_ms > maxwait\_ms\). If the lock is already held, the default 0 will cause the timeout_error to be raised immediately on a simultaneous access attempt
pulse_ms – milliseconds between subsequent attempts to acquire the lock
timeout_error – exception class raised when maxwait_ms exceeded. If provided, must be a subclass of Locked

Return type

Lock

A simple example that sleeps whenever the lock is acquired, in order to run out the 5-second clock on other waiting requests:

import asyncio
import random
import time

from ponty import get, render_json
from ponty.memo import locallock, Locked


mylock = locallock(maxwait_ms=5000)  # concurrent requests wait 5 seconds max


@get("/locktest")
@render_json
async def _locktest(_):
    secs = random.randint(1, 6)
    success = True

    try:
        async with mylock.lock("ponty"):  # "ponty" is the id of the protected resource
            await asyncio.sleep(secs)
    except Locked:
        success = False

    return {
        "success": success,
        "secs": secs,
    }

The five requests below are issued in parallel, so attempt to grab the lock at approximately the same time. Note how the first three acquire it at \(t = 0\), \(t = 1\), and \(t = 4\), respectively - all beneath the 5-second timeout - while the last two time out at roughly the 5-second mark. (With a larger timeout, they would have captured the lock at approx \(t = 8\) and \(t = 10\).)

$ for i in {1..5}; do curl localhost:8080/locktest & done
{"now": 1661271438402, "data": {"success": true, "secs": 1}, "elapsed": 1006}
{"now": 1661271438402, "data": {"success": true, "secs": 3}, "elapsed": 4049}
{"now": 1661271438402, "data": {"success": true, "secs": 4}, "elapsed": 8087}
{"now": 1661271438403, "data": {"success": false, "secs": 2}, "elapsed": 5232}
{"now": 1661271438404, "data": {"success": false, "secs": 5}, "elapsed": 5232}

Building blocks

Base classes for building a custom mutex.

class ponty.memo.SentinelStore

Container for asset id’s currently protected by a Lock.

Abstact Base Class. Lock expects the three abstract methods below - exists, add, remove - to return awaitables, whether they perform IO or not. (NB no context switch occurs if the method does not perform a blocking operation.) This constraint may be relieved in a future version.

abstract async add(key)

Take the lock on the asset.

Parameters: key (str) – unique id for the asset
Return type: None

abstract async exists(key)

Check if an asset is currently locked.

Parameters: key (str) – unique id for the asset
Returns: True if the asset is locked, False otherwise
Return type: bool

abstract async remove(key)

Release the lock on the asset.

Parameters: key (str) – unique id for the asset
Return type: None

class ponty.memo.Lock(sentinels, maxwait_ms=0, pulse_ms=100, timeout_error=<class 'ponty.memo.lock.base.Locked'>)

Basic mechanic for locking / antistampede.

Operates as a mandatory lock by default. Control blocking by specifying the maxwait_ms parameter.

Parameters

sentinels (SentinelStore) – set-like container, providing add/remove operations and existence checking. Must be an instance of SentinelStore
maxwait_ms (int) – total milliseconds to wait, before raising an exception with the caller. Raises timeout_error when \(n * pulse\_ms > maxwait\_ms\)
pulse_ms (int) – milliseconds between subsequent attempts to acquire the lock
timeout_error (type[Locked]) – exception class raised when maxwait_ms exceeded. If provided, must be a subclass of Locked

lock(key)

Context manager. Holds the lock on key until the block exits, or waits to acquire it until maxwait_ms milliseconds have elapsed.

Parameters: key (str) – unique identifier for the asset whose access requires synchronization
Return type: AsyncIterator[None]

timeout_error: alias of Locked

exception ponty.memo.Locked(*, text=None, body=None, **kw)

Returns a 409. Inherits PontyError.

Parameters

text (Optional[str]) –
body (Any) –

Footnotes

1: Under the hood, we’re using aiohttp.web.Application.cleanup_ctx for startup/cleanup handling.