apifrom.middleware
Middleware for APIFromAnything.
This module provides middleware for common tasks such as CORS, caching, rate limiting, security headers, authentication, and more.
Overview
Classes
AdvancedCacheMiddlewareBaseMiddlewareCORSMiddlewareCacheMiddlewareErrorHandlingMiddlewareHybridEvictionPolicyInMemoryRateLimitBackendLFUEvictionPolicyLRUEvictionPolicyMemoryCacheBackendMiddlewareRateLimitMiddlewareRedisCacheBackendSizeEvictionPolicyTTLEvictionPolicy
Classes
- AdvancedCacheMiddleware(cache_backend = None, ttl = 60, cache_methods = None, ignore_paths = None, vary_headers = None, cache_control_header = True, compress_responses = False, endpoint_ttls = None, auto_vary = True, invalidation_strategy = None):bases: CacheMiddleware
Advanced middleware for caching API responses.
This middleware extends the basic CacheMiddleware with additional features such as per-endpoint TTL, response compression, and automatic cache key generation.
Initialize the advanced cache middleware.
- param cache_backend:
The cache backend to use (defaults to MemoryCacheBackend)
- param ttl:
Default time-to-live in seconds
- param cache_methods:
Set of HTTP methods to cache (defaults to {“GET”})
- param ignore_paths:
Set of paths to exclude from caching
- param vary_headers:
Set of headers to include in the cache key
- param cache_control_header:
Whether to set Cache-Control headers
- param compress_responses:
Whether to compress responses before caching
- param endpoint_ttls:
Dictionary mapping endpoints to TTL values
- param auto_vary:
Whether to automatically determine vary headers
- param invalidation_strategy:
Strategy for cache invalidation
- apifrom.middleware._generate_cache_key(request)
Generate a cache key for a request.
- param request:
The request object
- returns:
A cache key string
- apifrom.middleware._get_endpoint_ttl(path)
Get the TTL for an endpoint.
- param path:
The endpoint path
- returns:
The TTL in seconds
- apifrom.middleware.clear()
Clear the cache.
- apifrom.middleware.get_stats()
Get cache statistics.
- returns:
A dictionary of cache statistics
- apifrom.middleware.invalidate(pattern)
Invalidate cache entries matching a pattern.
- param pattern:
The pattern to match
- apifrom.middleware.process_request(request, call_next)
- :async:
Process a request and potentially return a cached response.
- param request:
The request object
- param call_next:
The next middleware function
- returns:
The response object
- BaseMiddleware(**options):bases: abc.ABC
Base middleware class for APIFromAnything.
- apifrom.middleware.process_request(request)
- :abstractmethod:
- :async:
Process a request.
- param request:
The request to process.
- returns:
The processed request.
- apifrom.middleware.process_response(response)
- :abstractmethod:
- :async:
Process a response.
- param response:
The response to process.
- returns:
The processed response.
- CORSMiddleware(allow_origins = None, allow_methods = None, allow_headers = None, allow_credentials = False, expose_headers = None, max_age = 600, preflight_continue = False, options_success_status = 204):bases: apifrom.middleware.base.BaseMiddleware
Middleware for handling Cross-Origin Resource Sharing (CORS).
This middleware adds appropriate CORS headers to responses to allow controlled cross-origin requests to the API.
Initialize the CORS middleware.
- param allow_origins:
List of origins that are allowed to make cross-origin requests, or “*” to allow any origin
- param allow_methods:
List of HTTP methods that are allowed for cross-origin requests
- param allow_headers:
List of HTTP headers that can be used in cross-origin requests
- param allow_credentials:
Whether cookies and credentials can be included in cross-origin requests
- param expose_headers:
List of HTTP headers that can be exposed to the browser
- param max_age:
How long the results of a preflight request can be cached (in seconds)
- param preflight_continue:
Whether to continue processing after a preflight request
- param options_success_status:
Status code to use for successful OPTIONS requests
- apifrom.middleware._add_cors_headers(request, response)
Add CORS headers to a response.
- param request:
The request
- param response:
The response to add headers to
- apifrom.middleware._handle_preflight(request)
Handle a CORS preflight request.
- param request:
The preflight request
- returns:
A response for the preflight request, or None to continue processing
- apifrom.middleware._is_origin_allowed(origin)
Check if an origin is allowed.
- param origin:
The origin to check
- returns:
True if the origin is allowed, False otherwise
- apifrom.middleware.process_request(request)
- :async:
Process a request through the CORS middleware.
- param request:
The request to process
- returns:
The processed request
- apifrom.middleware.process_response(response)
- :async:
Process a response through the CORS middleware.
- param response:
The response to process
- returns:
The processed response
Base class for cache backends.
- apifrom.middleware.clear()
- :abstractmethod:
Clear the cache.
- apifrom.middleware.delete(key)
- :abstractmethod:
Delete a value from the cache.
- param key:
The cache key
- returns:
True if the key was deleted, False otherwise
- apifrom.middleware.get(key)
- :abstractmethod:
Get a value from the cache.
- param key:
The cache key
- returns:
The cached value, or None if not found
- apifrom.middleware.get_stats()
- :abstractmethod:
Get cache statistics.
- returns:
A dictionary of cache statistics
- apifrom.middleware.set(key, value, ttl=60)
- :abstractmethod:
Set a value in the cache.
- param key:
The cache key
- param value:
The value to cache
- param ttl:
Time-to-live in seconds
Cache control decorators for API endpoints.
This class provides decorators to control caching behavior for API endpoints. It includes decorators to cache responses, prevent caching, and invalidate cache entries.
Example
```python from apifrom import API, api from apifrom.middleware.cache_advanced import CacheControl
app = API()
@api(route=”/users/{user_id}”, method=”GET”) @CacheControl.cache(ttl=60, tags=[“user”]) def get_user(user_id: str):
# This response will be cached for 60 seconds return {“id”: user_id, “name”: “John”}
@api(route=”/users/{user_id}”, method=”PUT”) @CacheControl.invalidate([“user”]) def update_user(user_id: str, name: str):
# This will invalidate all cache entries with the “user” tag return {“id”: user_id, “name”: name}
@api(route=”/users/{user_id}/sensitive”, method=”GET”) @CacheControl.no_cache def get_sensitive_user_data(user_id: str):
# This response will not be cached return {“id”: user_id, “ssn”: “123-45-6789”}
- apifrom.middleware.cache(ttl=60, tags=None, dependencies=None)
- :staticmethod:
Decorator to cache the response of an API endpoint.
- param ttl:
Time to live in seconds
- param tags:
Tags to associate with the cache entry
- param dependencies:
Dependencies to associate with the cache entry
- returns:
A decorator function
- apifrom.middleware.invalidate(patterns)
- :staticmethod:
Decorator to invalidate cache entries matching the given patterns.
- param patterns:
Patterns to match cache keys
- returns:
A decorator function
- apifrom.middleware.no_cache(func)
- :staticmethod:
Decorator to prevent caching of an API endpoint.
- param func:
The function to decorate
- returns:
The decorated function
- CacheMiddleware(cache_backend = None, ttl = 60, methods = None, exclude_routes = None, vary_headers = None, key_prefix = 'apifrom-cache:'):bases: apifrom.middleware.base.BaseMiddleware
Middleware for caching API responses.
Initialize the cache middleware.
- param cache_backend:
The cache backend to use (defaults to MemoryCache)
- param ttl:
Default time-to-live in seconds for cached items
- param methods:
HTTP methods to cache (defaults to [“GET”])
- param exclude_routes:
Routes to exclude from caching
- param vary_headers:
Headers to include in the cache key
- param key_prefix:
Prefix for cache keys
- apifrom.middleware.__call__(scope, receive, send)
- :async:
ASGI callable.
- param scope:
The ASGI scope.
- param receive:
The ASGI receive function.
- param send:
The ASGI send function.
- apifrom.middleware._generate_cache_key(request)
Generate a cache key for a request.
- param request:
The request object
- returns:
The cache key
- apifrom.middleware._should_cache(request)
Determine if a request should be cached.
- param request:
The request object
- returns:
True if the request should be cached, False otherwise
- apifrom.middleware.process_request(request)
- :async:
Process a request through the cache middleware.
- param request:
The request object
- returns:
The request object
- apifrom.middleware.process_response(response)
- :async:
Process a response through the cache middleware.
- param response:
The response object
- returns:
The response object
- class apifrom.middleware.DependencyBasedInvalidation(cache_backend)[source]
- Parameters:
cache_backend (CacheBackend)
Dependency-based cache invalidation strategy.
This class provides a way to invalidate cache entries based on dependencies. Dependencies are relationships between cache entries, where invalidating one entry will also invalidate all entries that depend on it.
Example
```python # Create a cache backend cache_backend = MemoryCacheBackend()
# Create a dependency-based invalidation strategy invalidation = DependencyBasedInvalidation(cache_backend)
# Set a cache entry with dependencies cache_backend.set(“user:123”, {“name”: “John”}) invalidation.add_dependency(“user:123”, “users”)
# Set another cache entry with dependencies cache_backend.set(“post:456”, {“title”: “Hello”}) invalidation.add_dependency(“post:456”, “posts”) invalidation.add_dependency(“post:456”, “user:123”)
# Invalidate all cache entries that depend on “user:123” invalidation.invalidate(“user:123”) ```
Initialize the dependency-based invalidation strategy.
- param cache_backend:
The cache backend to use
- apifrom.middleware.add_dependencies(key, dependencies)
Add multiple dependencies to a cache entry.
- param key:
The cache key
- param dependencies:
List of dependency keys
- apifrom.middleware.add_dependency(key, dependency)
Add a dependency to a cache entry.
- param key:
The cache key
- param dependency:
The dependency key
- apifrom.middleware.get_dependencies(key)
Get all dependencies of a cache entry.
- param key:
The cache key
- returns:
A list of dependency keys
- apifrom.middleware.get_dependents(key)
Get all cache entries that depend on a key.
- param key:
The dependency key
- returns:
A list of cache keys
- apifrom.middleware.invalidate(key)
Invalidate a cache entry and all entries that depend on it.
- param key:
The key to invalidate
- apifrom.middleware.invalidate_dependency(dependency)
Invalidate all cache entries that depend on a specific dependency.
- param dependency:
The dependency key to invalidate
- ErrorHandlingMiddleware(debug = False, include_traceback = False, include_exception_class = False, log_exceptions = True, json_encoder = None, **kwargs):bases: apifrom.middleware.base.Middleware
Middleware for handling errors in API requests.
This middleware catches exceptions and returns appropriate error responses.
Initialize the error handling middleware.
- param debug:
Whether to include debug information in error responses
- param include_traceback:
Whether to include tracebacks in debug mode
- param include_exception_class:
Whether to include exception class names
- param log_exceptions:
Whether to log exceptions
- param json_encoder:
A custom JSON encoder for error responses
- param **kwargs:
Additional options for the base middleware
- apifrom.middleware._create_error_response(message, status_code, error_code, exception, details=None)
Create an error response.
- param message:
The error message
- param status_code:
The HTTP status code
- param error_code:
The error code
- param exception:
The original exception
- param details:
Additional error details
- returns:
An API response
- apifrom.middleware._find_handler(exception)
Find the appropriate handler for an exception.
- param exception:
The exception to handle
- returns:
A handler function
- apifrom.middleware._handle_api_error(exception)
Handle an API error.
- param exception:
The API error
- returns:
An API response
- apifrom.middleware._setup_default_handlers()
Set up default exception handlers.
- apifrom.middleware.add_exception_handler(exception_class, handler)
Add a custom exception handler.
- param exception_class:
The exception class to handle
- param handler:
A function that converts the exception to a response
- apifrom.middleware.dispatch(request, call_next)
- :async:
Dispatch a request, catching and handling any exceptions.
- param request:
The request to process
- param call_next:
The next middleware or route handler
- returns:
The response
- apifrom.middleware.process_request(request)
- :async:
Process a request.
- param request:
The request to process
- returns:
The processed request
- HybridEvictionPolicy:bases: CacheEvictionPolicy
Hybrid eviction policy combining multiple factors.
- apifrom.middleware.select_items_to_evict(items, target_size)
Select items to evict based on a hybrid policy.
- param items:
The list of cache items
- param target_size:
The target number of items to keep
- returns:
A list of items to evict
- InMemoryRateLimitBackend:bases: RateLimitBackend
In-memory implementation of RateLimitBackend.
This backend stores rate limit data in memory.
Initialize a new InMemoryRateLimitBackend instance.
- apifrom.middleware._cleanup()
Clean up expired keys.
- apifrom.middleware.get(key)
Get rate limit data for a key.
- param key:
The key to get data for
- returns:
The rate limit data, or None if not found
- apifrom.middleware.increment(key, amount=1, ttl=None)
Increment rate limit counter for a key.
- param key:
The key to increment
- param amount:
The amount to increment by
- param ttl:
Time to live in seconds
- returns:
The new counter value
- apifrom.middleware.reset(key)
Reset rate limit data for a key.
- param key:
The key to reset
- apifrom.middleware.set(key, value, ttl=None)
Set rate limit data for a key.
- param key:
The key to set data for
- param value:
The data to set
- param ttl:
Time to live in seconds
- LFUEvictionPolicy:bases: CacheEvictionPolicy
Least Frequently Used (LFU) eviction policy.
- apifrom.middleware.select_items_to_evict(items, target_size)
Select items to evict based on the LFU policy.
- param items:
The list of cache items
- param target_size:
The target number of items to keep
- returns:
A list of items to evict
- LRUEvictionPolicy:bases: CacheEvictionPolicy
Least Recently Used (LRU) eviction policy.
- apifrom.middleware.select_items_to_evict(items, target_size)
Select items to evict based on the LRU policy.
- param items:
The list of cache items
- param target_size:
The target number of items to keep
- returns:
A list of items to evict
- MemoryCacheBackend(max_items = 1000, max_size_bytes = 50 * 1024 * 1024, eviction_policy = None):bases: CacheBackend
In-memory cache backend.
Initialize the memory cache backend.
- param max_items:
Maximum number of items to store
- param max_size_bytes:
Maximum cache size in bytes
- param eviction_policy:
The eviction policy to use
- apifrom.middleware._cleanup_if_needed(force=False)
Clean up expired items if needed.
- param force:
Whether to force cleanup regardless of the time since last cleanup
- apifrom.middleware._evict_if_needed()
Evict items if the cache exceeds its limits.
- apifrom.middleware.clear()
Clear the cache.
- apifrom.middleware.delete(key)
Delete a value from the cache.
- param key:
The cache key
- returns:
True if the key was deleted, False otherwise
- apifrom.middleware.get(key)
Get a value from the cache.
- param key:
The cache key
- returns:
The cached value, or None if not found
- apifrom.middleware.get_stats()
Get cache statistics.
- returns:
A dictionary of cache statistics
- apifrom.middleware.set(key, value, ttl=60)
Set a value in the cache.
- param key:
The cache key
- param value:
The value to cache
- param ttl:
Time-to-live in seconds
- Middleware(app, dispatch = None):bases: starlette.middleware.base.BaseHTTPMiddleware, BaseMiddleware
Middleware class for APIFromAnything.
This class implements the BaseMiddleware interface and extends Starlette’s BaseHTTPMiddleware to provide a middleware component that can be used with Starlette.
Initialize a new BaseMiddleware instance.
- param **options:
Options for the middleware.
- apifrom.middleware.dispatch(request, call_next)
- :async:
Dispatch a request.
This method is called by Starlette to process a request.
- param request:
The request to process.
- param call_next:
A function to call the next middleware or route handler.
- returns:
The response.
- apifrom.middleware.process_request(request)
- :abstractmethod:
- :async:
Process a request before it is handled by the API.
- param request:
The request to process.
- returns:
The processed request.
- apifrom.middleware.process_response(response)
- :async:
Process a response after it is handled by the API.
- param response:
The response to process.
- returns:
The processed response.
Base class for rate limit backends.
Rate limit backends are responsible for storing and retrieving rate limit data.
- apifrom.middleware.get(key)
- :abstractmethod:
Get rate limit data for a key.
- param key:
The key to get data for
- returns:
The rate limit data
- apifrom.middleware.increment(key, amount=1, ttl=None)
- :abstractmethod:
Increment rate limit counter for a key.
- param key:
The key to increment
- param amount:
The amount to increment by
- param ttl:
Time to live in seconds
- returns:
The new counter value
- apifrom.middleware.reset(key)
- :abstractmethod:
Reset rate limit data for a key.
- param key:
The key to reset
- apifrom.middleware.set(key, value, ttl=None)
- :abstractmethod:
Set rate limit data for a key.
- param key:
The key to set data for
- param value:
The data to set
- param ttl:
Time to live in seconds
- RateLimitMiddleware(limiter, key_func = None, exclude_routes = None, headers_enabled = True):bases: apifrom.middleware.base.BaseMiddleware
Middleware for rate limiting API requests.
Initialize the rate limit middleware.
- param limiter:
The rate limiter to use
- param key_func:
Function to extract the rate limit key from a request
- param exclude_routes:
Routes to exclude from rate limiting
- param headers_enabled:
Whether to include rate limit headers in responses
- apifrom.middleware.__call__(scope, receive, send)
- :async:
ASGI callable.
- param scope:
The ASGI scope.
- param receive:
The ASGI receive function.
- param send:
The ASGI send function.
- apifrom.middleware._add_rate_limit_headers(response, limit_info)
Add rate limit headers to a response.
- param response:
The response object
- param limit_info:
Rate limit information
- apifrom.middleware._default_key_func(request)
Default function to extract the rate limit key from a request.
- param request:
The request object
- returns:
The rate limit key
- apifrom.middleware._should_limit(request)
Determine if a request should be rate limited.
- param request:
The request object
- returns:
True if the request should be rate limited, False otherwise
- apifrom.middleware.process_request(request)
- :async:
Process a request through the rate limit middleware.
- param request:
The request object
- returns:
The request object
- apifrom.middleware.process_response(response)
- :async:
Process a response through the rate limit middleware.
- param response:
The response object
- returns:
The response object
- RedisCacheBackend(redis_url = 'redis://localhost:6379/0', prefix = 'apifrom:', serializer = None, deserializer = None):bases: CacheBackend
Redis cache backend.
Initialize the Redis cache backend.
- param redis_url:
The Redis connection URL
- param prefix:
The key prefix to use
- param serializer:
Function to serialize values to bytes
- param deserializer:
Function to deserialize bytes to values
- apifrom.middleware._make_key(key)
Create a prefixed key.
- param key:
The original key
- returns:
The prefixed key
- apifrom.middleware.clear()
Clear the cache.
- apifrom.middleware.delete(key)
Delete a value from the cache.
- param key:
The cache key
- returns:
True if the key was deleted, False otherwise
- apifrom.middleware.get(key)
Get a value from the cache.
- param key:
The cache key
- returns:
The cached value, or None if not found
- apifrom.middleware.get_stats()
Get cache statistics.
- returns:
A dictionary of cache statistics
- apifrom.middleware.set(key, value, ttl=60)
Set a value in the cache.
- param key:
The cache key
- param value:
The value to cache
- param ttl:
Time-to-live in seconds
- SizeEvictionPolicy:bases: CacheEvictionPolicy
Size-based eviction policy.
- apifrom.middleware.select_items_to_evict(items, target_size)
Select items to evict based on size.
- param items:
The list of cache items
- param target_size:
The target number of items to keep
- returns:
A list of items to evict
- TTLEvictionPolicy:bases: CacheEvictionPolicy
Time-To-Live (TTL) eviction policy.
- apifrom.middleware.select_items_to_evict(items, target_size)
Select items to evict based on the TTL policy.
- param items:
The list of cache items
- param target_size:
The target number of items to keep
- returns:
A list of items to evict
- class apifrom.middleware.TagBasedInvalidation(cache_backend)[source]
- Parameters:
cache_backend (CacheBackend)
Tag-based cache invalidation strategy.
This class provides a way to invalidate cache entries based on tags. Tags are arbitrary strings that can be associated with cache entries. When a tag is invalidated, all cache entries associated with that tag are also invalidated.
Example
```python # Create a cache backend cache_backend = MemoryCacheBackend()
# Create a tag-based invalidation strategy invalidation = TagBasedInvalidation(cache_backend)
# Set a cache entry with tags cache_backend.set(“user:123”, {“name”: “John”}) invalidation.tag(“user:123”, [“user”, “user:123”])
# Invalidate all cache entries with the “user” tag invalidation.invalidate_tag(“user”) ```
Initialize the tag-based invalidation strategy.
- param cache_backend:
The cache backend to use
- apifrom.middleware.get_keys_for_tag(tag)
Get all cache keys associated with a tag.
- param tag:
The tag to get keys for
- returns:
A list of cache keys
- apifrom.middleware.invalidate_tag(tag)
Invalidate all cache entries associated with a tag.
- param tag:
The tag to invalidate
- apifrom.middleware.invalidate_tags(tags)
Invalidate all cache entries associated with any of the given tags.
- param tags:
The tags to invalidate
- apifrom.middleware.tag(key, tags)
Tag a cache entry with one or more tags.
- param key:
The cache key
- param tags:
The tags to associate with the key