Caching

Caching infrastructure for performance optimization.

Module Contents

Page caching module for Redis-backed response caching.

class AdminNoCacheMiddleware

Bases: AbstractMiddleware

Middleware to prevent admin paths from being cached by CDN or browsers.

Admin pages should never be cached to ensure users always see fresh content and prevent sensitive data from being stored in shared caches.

scopes: Scopes = {ScopeType.HTTP}

class CacheControlMiddleware

Bases: AbstractMiddleware

Middleware to add Cache-Control headers based on path patterns.

Applies appropriate caching directives for different content types: - Static files: public, max-age=31536000 (1 year) - API responses: no-store (unless route specifies otherwise) - Pages: public, max-age=300 (5 minutes, overridable by route cache)

API_PATHS = ('/api/',)

NO_CACHE_PATHS = ('/admin/', '/auth/')

PAGE_MAX_AGE = 300

STATIC_MAX_AGE = 31536000

STATIC_PATHS = ('/static/', '/media/', '/assets/')

scopes: Scopes = {ScopeType.HTTP}

class PageCacheService

Bases: object

Service for managing page cache entries.

Provides methods to invalidate cached pages when content changes, ensuring users always see the most up-to-date content after edits.

Example

>>> cache_service = PageCacheService(redis_client)
>>> await cache_service.invalidate_page("/about/history")
>>> await cache_service.invalidate_all_pages()

__init__(redis)

Initialize the page cache service.

Parameters:

redis (Redis) – Async Redis client instance.

async get_cache_stats()

Get statistics about cached pages.

Return type:

dict[str, int]

Returns:

Dictionary with cache statistics including count and memory usage.

async invalidate_all_pages()

Invalidate all cached pages using SCAN + DEL pattern.

Uses SCAN to avoid blocking Redis with large KEYS operations.

Return type:

int

Returns:

Number of cache entries deleted.

Example

>>> deleted = await cache_service.invalidate_all_pages()
>>> print(f"Cleared {deleted} cached pages")

async invalidate_page(path)

Invalidate the cache for a specific page path.

Parameters:

path (str) – The URL path of the page to invalidate.

Return type:

bool

Returns:

True if the cache entry was deleted, False if it didn’t exist.

Example

>>> await cache_service.invalidate_page("/about/history")
True

async invalidate_page_by_id(page_id, path=None)

Invalidate the cache for a page by ID.

If path is provided, uses it directly. Otherwise, this method cannot determine the path and returns False.

Parameters:

• page_id (UUID) – The UUID of the page.

• path (str | None) – Optional path of the page (for efficiency).

Return type:

bool

Returns:

True if cache was invalidated, False otherwise.

class SurrogateKeyMiddleware

Bases: AbstractMiddleware

Middleware to add Surrogate-Key headers for CDN cache purging.

Fastly and similar CDNs use Surrogate-Key headers to enable targeted cache invalidation. This middleware adds a global key to all responses and preserves any route-specific keys.

Example

A response with Surrogate-Key: page-123 will be sent as: Surrogate-Key: pydotorg-app page-123

This allows purging all cached content with the global key, or specific content with route-specific keys.

__init__(app, surrogate_key='pydotorg-app')

Initialize the middleware.

Parameters:

• app (Callable[[Union[HTTPScope, WebSocketScope], Callable[..., Awaitable[Union[HTTPRequestEvent, HTTPDisconnectEvent, WebSocketConnectEvent, WebSocketReceiveEvent, WebSocketDisconnectEvent]]], Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent]], Awaitable[None]]], Awaitable[None]]) – The ASGI application.

• surrogate_key (str) – Global surrogate key for all responses.

scopes: Scopes = {ScopeType.HTTP}

create_cache_middleware_stack(*, enable_surrogate_keys=True, surrogate_key='pydotorg-app')

Create the cache middleware stack for the application.

Parameters:

• enable_surrogate_keys (bool) – Enable Surrogate-Key headers for CDN purging.

• surrogate_key (str) – Global surrogate key value.

Return type:

list[type[AbstractMiddleware]]

Returns:

List of middleware classes to add to the application.

create_response_cache_config()

Create response cache configuration for the application.

Configures caching with: - Default TTL of 60 seconds for general routes - Redis store backend (configured separately) - Custom key builder for page routes

Note

The Redis store must be configured in the Litestar app’s stores parameter:

>>> from litestar.stores.redis import RedisStore
>>> redis_store = RedisStore.with_client(url=settings.redis_url)
>>> app = Litestar(stores={"response_cache": redis_store}, ...)

Returns:

Configured response cache instance.

Return type:

ResponseCacheConfig

page_cache_key_builder(request)

Build a cache key for page requests based on URL path.

Creates a unique cache key for each page URL, ensuring that: - Different pages have different cache keys - Query parameters are included in the key - The key is a fixed-length hash for consistent storage

Parameters:

request (Request) – The Litestar request object.

Return type:

str

Returns:

A hex digest of the cache key suitable for Redis.

Example

>>> # For URL /about/history?section=timeline
>>> key = page_cache_key_builder(request)
>>> key  # 'page:a1b2c3d4e5f6...'

Configuration

Cache configuration and settings.

Response cache configuration for Litestar.

This module provides Redis-backed response caching with configurable TTLs for different content types. Pages are cached to reduce database load and improve response times.

Example

>>> from pydotorg.core.cache import create_response_cache_config
>>> cache_config = create_response_cache_config()
>>> app = Litestar(
...     response_cache_config=cache_config,
...     stores={"response_cache": redis_store},
... )

page_cache_key_builder(request)

Build a cache key for page requests based on URL path.

Creates a unique cache key for each page URL, ensuring that: - Different pages have different cache keys - Query parameters are included in the key - The key is a fixed-length hash for consistent storage

Parameters:

request (Request) – The Litestar request object.

Return type:

str

Returns:

A hex digest of the cache key suitable for Redis.

Example

>>> # For URL /about/history?section=timeline
>>> key = page_cache_key_builder(request)
>>> key  # 'page:a1b2c3d4e5f6...'

create_response_cache_config()

Create response cache configuration for the application.

Configures caching with: - Default TTL of 60 seconds for general routes - Redis store backend (configured separately) - Custom key builder for page routes

Note

The Redis store must be configured in the Litestar app’s stores parameter:

>>> from litestar.stores.redis import RedisStore
>>> redis_store = RedisStore.with_client(url=settings.redis_url)
>>> app = Litestar(stores={"response_cache": redis_store}, ...)

Returns:

Configured response cache instance.

Return type:

ResponseCacheConfig

Middleware

Cache middleware for response caching.

Cache control middleware for HTTP headers.

Provides middleware for: - Admin no-caching (private Cache-Control) - Surrogate-Key headers for CDN purging (Fastly) - Cache-Control headers for various content types

class AdminNoCacheMiddleware

Bases: AbstractMiddleware

Middleware to prevent admin paths from being cached by CDN or browsers.

Admin pages should never be cached to ensure users always see fresh content and prevent sensitive data from being stored in shared caches.

scopes: Scopes = {ScopeType.HTTP}

class SurrogateKeyMiddleware

Bases: AbstractMiddleware

Middleware to add Surrogate-Key headers for CDN cache purging.

Fastly and similar CDNs use Surrogate-Key headers to enable targeted cache invalidation. This middleware adds a global key to all responses and preserves any route-specific keys.

Example

A response with Surrogate-Key: page-123 will be sent as: Surrogate-Key: pydotorg-app page-123

This allows purging all cached content with the global key, or specific content with route-specific keys.

scopes: Scopes = {ScopeType.HTTP}

__init__(app, surrogate_key='pydotorg-app')

Initialize the middleware.

Parameters:

• app (Callable[[Union[HTTPScope, WebSocketScope], Callable[..., Awaitable[Union[HTTPRequestEvent, HTTPDisconnectEvent, WebSocketConnectEvent, WebSocketReceiveEvent, WebSocketDisconnectEvent]]], Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent]], Awaitable[None]]], Awaitable[None]]) – The ASGI application.

• surrogate_key (str) – Global surrogate key for all responses.

class CacheControlMiddleware

Bases: AbstractMiddleware

Middleware to add Cache-Control headers based on path patterns.

Applies appropriate caching directives for different content types: - Static files: public, max-age=31536000 (1 year) - API responses: no-store (unless route specifies otherwise) - Pages: public, max-age=300 (5 minutes, overridable by route cache)

scopes: Scopes = {ScopeType.HTTP}

STATIC_PATHS = ('/static/', '/media/', '/assets/')

API_PATHS = ('/api/',)

NO_CACHE_PATHS = ('/admin/', '/auth/')

STATIC_MAX_AGE = 31536000

PAGE_MAX_AGE = 300

create_cache_middleware_stack(*, enable_surrogate_keys=True, surrogate_key='pydotorg-app')

Create the cache middleware stack for the application.

Parameters:

• enable_surrogate_keys (bool) – Enable Surrogate-Key headers for CDN purging.

• surrogate_key (str) – Global surrogate key value.

Return type:

list[type[AbstractMiddleware]]

Returns:

List of middleware classes to add to the application.

Service

Cache service for application-level caching operations.

Page cache service for manual cache management.

This module provides a service for programmatic cache invalidation, used when pages are updated, published, or unpublished to ensure users see fresh content.

class PageCacheService

Bases: object

Service for managing page cache entries.

Provides methods to invalidate cached pages when content changes, ensuring users always see the most up-to-date content after edits.

Example

>>> cache_service = PageCacheService(redis_client)
>>> await cache_service.invalidate_page("/about/history")
>>> await cache_service.invalidate_all_pages()

__init__(redis)

Initialize the page cache service.

Parameters:

redis (Redis) – Async Redis client instance.

async invalidate_page(path)

Invalidate the cache for a specific page path.

Parameters:

path (str) – The URL path of the page to invalidate.

Return type:

bool

Returns:

True if the cache entry was deleted, False if it didn’t exist.

Example

>>> await cache_service.invalidate_page("/about/history")
True

async invalidate_page_by_id(page_id, path=None)

Invalidate the cache for a page by ID.

If path is provided, uses it directly. Otherwise, this method cannot determine the path and returns False.

Parameters:

• page_id (UUID) – The UUID of the page.

• path (str | None) – Optional path of the page (for efficiency).

Return type:

bool

Returns:

True if cache was invalidated, False otherwise.

async invalidate_all_pages()

Invalidate all cached pages using SCAN + DEL pattern.

Uses SCAN to avoid blocking Redis with large KEYS operations.

Return type:

int

Returns:

Number of cache entries deleted.

Example

>>> deleted = await cache_service.invalidate_all_pages()
>>> print(f"Cleared {deleted} cached pages")

async get_cache_stats()

Get statistics about cached pages.

Return type:

dict[str, int]

Returns:

Dictionary with cache statistics including count and memory usage.