Skip to content

Metrics

Application Kit provides request counting and metrics collection through decorators and dependencies.

Overview

The metrics system tracks request counts per endpoint and queues them to Redis for processing. This enables:

  • Usage tracking per project
  • Billing and quota management

Not for analytics

This metrics system is designed for billing and quota enforcement, not user behavior analytics. It counts API calls per project to track usage against subscription limits.

Usage

Use the AddJobToMetrics dependency:

"""FastAPI metrics example."""

from functools import partial

from fastapi import APIRouter, Depends

from application_kit.fastapi.fastapi import get_fastapi_app
from application_kit.fastapi.metrics import AddJobToMetrics

app = get_fastapi_app("api")
router = APIRouter()
app.include_router(router)

# Create a partial for your service
add_stores_metric = partial(AddJobToMetrics, "STORES")


@router.get(
    "/search",
    dependencies=[Depends(add_stores_metric("search"))],
)
async def search() -> dict[str, list[str]]:
    return {"results": []}

Note

The AddJobToMetrics dependency uses FastAPI's BackgroundTasks to send the job to Redis after the response is sent.

Use the @count_request decorator:

"""Django metrics example."""

from django.http import HttpRequest, JsonResponse

from application_kit.django.decorators import authenticate_key, count_request


@authenticate_key()
@count_request("search", "STORES")
def search_view(request: HttpRequest) -> JsonResponse:
    return JsonResponse({"results": []})


@authenticate_key()
@count_request("search", "STORES")
async def async_search_view(request: HttpRequest) -> JsonResponse:
    return JsonResponse({"results": []})

Use the Django @count_request decorator with @chain():

"""Django Ninja metrics example."""

from django.http import HttpRequest

from application_kit.django.decorators import authenticate_key, count_request
from application_kit.shinobi.api import WoosmapApi
from application_kit.shinobi.authentication import PublicKeyAuth
from application_kit.shinobi.decorators import chain, terminate

api = WoosmapApi(description="My API")


@api.get("/search", auth=[PublicKeyAuth()])
@chain(authenticate_key(), count_request("search", "STORES"))
@terminate()
def search(request: HttpRequest) -> dict[str, list[str]]:
    return {"results": []}

Parameters

Parameter Type Description
product str Product category (first argument to partial)
endpoint_name str Identifier for the endpoint
Parameter Type Description
endpoint_name str Identifier for the endpoint (e.g., "search")
product str Product category (e.g., "STORES", "MAPS")

Configuration

Add the metrics_queue Redis dependency to your application.json:

{
  "dependencies": {
    "databases": [
      {
        "name": "metrics_queue",
        "type": "redis",
        "extras": {"queue_prefix": "counter"}
      }
    ]
  }
}

The queue_prefix determines the Redis key prefix for the metrics queue.

How It Works

  1. When a request is processed, the decorator/dependency creates a job with:

    • Organization and project IDs
    • Endpoint name
    • Product category
    • Timestamp

  2. The job is pushed to a Redis queue (using the queue_prefix)

  3. A background worker processes the queue and aggregates metrics

API Reference

application_kit.fastapi.metrics

AddJobToMetrics 

AddJobToMetrics(product, kind)

AddJobToMetrics dependency

Source code in application_kit/fastapi/metrics.py 
13
14
15
def __init__(self, product: str, kind: str) -> None:
    self.product = product
    self.kind = kind

application_kit.django.decorators

count_request 

count_request(request_name, product, name_lambda=None)

Counts the request, must be placed after authenticate_key or authenticate_user decorator.

Source code in application_kit/django/decorators.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def count_request(
    request_name: str,
    product: str,
    name_lambda: CounterNameLambda | None = None,
) -> Callable[[T], T]:
    """Counts the request, must be placed after `authenticate_key` or `authenticate_user` decorator."""

    def _count_request(kwargs: Mapping[str, Any], request: HttpRequest) -> None:
        project_token = get_project_token(request)
        if project_token is not None:
            instance = project_token.instance
            kind = instance.kind
            request_kind = name_lambda(product, "", kind, **kwargs) if name_lambda else request_name
            add_job_to_queue_for_token(project_token, product, request_kind)
        else:
            warnings.warn(
                "Request was not authenticated, count_request decorator should be placed "
                "after a permission_classes decorator.",
                RuntimeWarning,
                stacklevel=2,
            )

    def wrapped(
        func: T,
    ) -> T:
        if is_coroutine_function(func):

            @wraps(func)
            async def wrapper(
                request: HttpRequest,
                *args: Any,
                **kwargs: Any,
            ) -> HttpResponseBase:
                _count_request(kwargs, request)

                return await func(request, *args, **kwargs)

            return wrapper  # type: ignore

        elif not_is_coroutine_function(func):

            @wraps(func)
            def wrapper(
                request: HttpRequest,
                *args: Any,
                **kwargs: Any,
            ) -> HttpResponseBase:
                _count_request(kwargs, request)

                return func(request, *args, **kwargs)

            return wrapper  # type: ignore
        else:
            raise RuntimeError(";_;")

    return wrapped