Async Patterns

Application Kit supports both synchronous and asynchronous execution patterns. This guide covers async-specific features and best practices for FastAPI and async Django views.

Async FastAPI Dependencies

FastAPI dependencies in Application Kit use async/await for non-blocking execution. The AuthenticateKey dependency calls the authentication service in a thread pool to avoid blocking:

from fastapi.concurrency import run_in_threadpool

# Inside FastAPIAuthenticateBase.authenticate()
project_token, _ = await run_in_threadpool(
    self.get_project_token,
    current_product,
    headers,
    key_token,
    kind,
    request_method,
    port,
)

This pattern is used because the underlying HTTP client uses synchronous libraries. The run_in_threadpool() helper runs blocking code in a separate thread, preventing the async event loop from being blocked.

Async Redis

For FastAPI applications, use get_async_redis_instance() to obtain an async Redis client:

from application_kit.settings import get_async_redis_instance

redis = get_async_redis_instance("ratelimit")
value = await redis.get("my_key")
await redis.set("my_key", "value")

Connection Pooling

Async Redis connections use a ConnectionPool that is cached per dependency name. The first call creates the pool, and subsequent calls reuse it:

# First call - creates connection pool
redis1 = get_async_redis_instance("ratelimit")

# Second call - reuses connection pool
redis2 = get_async_redis_instance("ratelimit")

# Different dependency - new pool
redis3 = get_async_redis_instance("authentication_cache")

Lua Scripts

Register and execute Lua scripts asynchronously for atomic operations:

from application_kit.ratelimit import PROJECT_RATE_LIMITER_LUA

redis = get_async_redis_instance("ratelimit")
script = redis.register_script(PROJECT_RATE_LIMITER_LUA)
result = await script([key, override_key], [max_requests, expiry])

Async Rate Limiting

All rate limit override functions in application_kit.fastapi.ratelimit are async:

from application_kit.fastapi.ratelimit import (
    set_rate_limit_override,
    get_rate_limit_override,
    delete_rate_limit_override,
    list_rate_limit_overrides,
    clear_project_rate_limit_overrides,
)

# All functions require await
await set_rate_limit_override(redis, org_id, project_id, "/path", 100, 60)
override = await get_rate_limit_override(redis, org_id, project_id, "/path")
overrides = await list_rate_limit_overrides(redis, org_id, project_id)

The ProjectRateLimiter dependency handles async execution internally.

Django Async Views

Application Kit decorators automatically detect async views using inspect.iscoroutinefunction():

from django.http import JsonResponse
from application_kit.django.decorators import authenticate_key

# Sync view - works normally
@authenticate_key()
def sync_view(request):
    return JsonResponse({"status": "ok"})

# Async view - decorator adapts automatically
@authenticate_key()
async def async_view(request):
    return JsonResponse({"status": "ok"})

The decorator creates appropriate wrapper functions for each case, ensuring proper async/await handling.

FastAPI Lifespan Management

For applications requiring startup/shutdown logic, use FastAPI's lifespan context manager:

from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI
from application_kit.fastapi.fastapi import get_fastapi_app
from application_kit.settings import get_async_redis_instance

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    # Startup: initialize connections
    redis = get_async_redis_instance("cache")
    await redis.ping()  # Verify connection

    yield  # Application runs

    # Shutdown: cleanup (Redis pools managed by application_kit)

app = get_fastapi_app("myservice", lifespan=lifespan)

SQLAlchemy Async

For FastAPI applications using SQLAlchemy with async support:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from application_kit.settings import get_database_url

def get_async_engine():
    db_url = get_database_url("default")
    if db_url is None:
        raise RuntimeError("Database not configured")
    # Use asyncpg driver for async PostgreSQL
    async_url = f"postgresql+asyncpg://{db_url.user}:{db_url.password}@{db_url.host}:{db_url.port}/{db_url.name}"
    return create_async_engine(async_url, pool_pre_ping=True)

engine = get_async_engine()
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_session():
    async with AsyncSessionLocal() as session:
        yield session

Usage in endpoints:

from fastapi import Depends

@router.get("/items")
async def list_items(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(Item))
    return result.scalars().all()

Best Practices

Use async Redis for FastAPI: Always use get_async_redis_instance() in FastAPI to avoid blocking the event loop.

Thread pool for blocking calls: When integrating with synchronous libraries, wrap calls in run_in_threadpool():

from fastapi.concurrency import run_in_threadpool
result = await run_in_threadpool(blocking_function, arg1, arg2)

Connection pool reuse: Application Kit caches connection pools automatically. Don't create multiple instances for the same dependency.

Decorator order matters: Place authentication decorators before rate limiting to ensure the project token is available:

@router.get(
    "/endpoint",
    dependencies=[
        Depends(AuthenticateKey()),      # First: authenticate
        Depends(ProjectRateLimiter()),   # Second: rate limit (needs project_token)
        Depends(AddJobToMetrics()),      # Third: count request
    ],
)

Test with pytest-asyncio: When testing async code, use @pytest.mark.asyncio:

import pytest

@pytest.mark.asyncio
async def test_async_endpoint():
    result = await some_async_function()
    assert result is not None