HTTP Cache#

Collect metrics based on prometheus.

class blacksmith.middleware._async.http_cache.AsyncAbstractCache#

Abstract Redis Client.

abstract async initialize() → None#: Initialize the cache

abstract async get(key: str) → Optional[str]#: Get a value from redis

abstract async set(key: str, val: str, ex: datetime.timedelta) → None#: Get a value from redis

class blacksmith.middleware._async.http_cache.AsyncHTTPCacheMiddleware(cache: blacksmith.middleware._async.http_cache.AsyncAbstractCache, metrics: Optional[blacksmith.domain.model.middleware.prometheus.PrometheusMetrics] = None, policy: blacksmith.domain.model.middleware.http_cache.AbstractCachePolicy = <blacksmith.domain.model.middleware.http_cache.CacheControlPolicy object>, serializer: Type[blacksmith.domain.model.middleware.http_cache.AbstractSerializer] = <class 'blacksmith.domain.model.middleware.http_cache.JsonSerializer'>)#

Http Cache Middleware based on Cache-Control and redis.

__init__(cache: blacksmith.middleware._async.http_cache.AsyncAbstractCache, metrics: Optional[blacksmith.domain.model.middleware.prometheus.PrometheusMetrics] = None, policy: blacksmith.domain.model.middleware.http_cache.AbstractCachePolicy = <blacksmith.domain.model.middleware.http_cache.CacheControlPolicy object>, serializer: Type[blacksmith.domain.model.middleware.http_cache.AbstractSerializer] = <class 'blacksmith.domain.model.middleware.http_cache.JsonSerializer'>) → None#

async initialize() → None#

Asynchronous initialization of a middleware.

For instance, used to initialize connection to storage backend.

async cache_response(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest, resp: blacksmith.domain.model.http.HTTPResponse) → bool#

async get_from_cache(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest) → Optional[blacksmith.domain.model.http.HTTPResponse]#

__call__(next: blacksmith.domain.typing.AsyncMiddleware) → blacksmith.domain.typing.AsyncMiddleware#: Call self as a function.

observe_cache_hit(client_name: str, method: str, path: str, status_code: int, latency: float) → None#

inc_cache_miss(client_name: str, cachable_state: typing_extensions.Literal[uncachable_request, uncachable_response, cached], method: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], path: str, status_code: int) → None#