Cache control parsing function

fast_cache_middleware/_helpers.py

@@ -1,4 +1,7 @@
+from hashlib import blake2b
+
 from fastapi import FastAPI, routing
+from starlette.requests import Request
 
 from .depends import CacheConfig
 
@@ -26,3 +29,31 @@ def set_cache_age_in_openapi_schema(app: FastAPI) -> None:
 
     app.openapi_schema = openapi_schema
     return None
+
+
+def generate_key(request: Request) -> str:
+    """Generates fast unique key for caching HTTP request.
+
+    Args:
+        request: Starlette Request object.
+
+    Returns:
+        str: Unique key for caching, based on request method and path.
+        Uses fast blake2b hashing algorithm.
+
+    Note:
+        Does not consider scheme and host, as requests usually go to the same host.
+        Only considers method, path and query parameters for maximum performance.
+    """
+    # Get only necessary components from scope
+    scope = request.scope
+    url = scope["path"]
+    if scope["query_string"]:
+        url += f"?{scope['query_string'].decode('ascii')}"
+
+    # Use fast blake2b algorithm with minimal digest size
+    key = blake2b(digest_size=8)
+    key.update(request.method.encode())
+    key.update(url.encode())
+
+    return key.hexdigest()

fast_cache_middleware/controller.py

@@ -1,51 +1,23 @@
 import http
 import logging
 import re
-from hashlib import blake2b
 from typing import Optional
 
 from starlette.concurrency import run_in_threadpool
 from starlette.requests import Request
 from starlette.responses import Response
 from starlette.routing import is_async_callable
 
+from ._helpers import generate_key
 from .exceptions import FastCacheMiddlewareError
-from .schemas import CacheConfiguration
+from .schemas import CacheConfiguration, CacheControlDirectives
 from .storages import BaseStorage
 
 logger = logging.getLogger(__name__)
 
 KNOWN_HTTP_METHODS = [method.value for method in http.HTTPMethod]
 
 
-def generate_key(request: Request) -> str:
-    """Generates fast unique key for caching HTTP request.
-
-    Args:
-        request: Starlette Request object.
-
-    Returns:
-        str: Unique key for caching, based on request method and path.
-        Uses fast blake2b hashing algorithm.
-
-    Note:
-        Does not consider scheme and host, as requests usually go to the same host.
-        Only considers method, path and query parameters for maximum performance.
-    """
-    # Get only necessary components from scope
-    scope = request.scope
-    url = scope["path"]
-    if scope["query_string"]:
-        url += f"?{scope['query_string'].decode('ascii')}"
-
-    # Use fast blake2b algorithm with minimal digest size
-    key = blake2b(digest_size=8)
-    key.update(request.method.encode())
-    key.update(url.encode())
-
-    return key.hexdigest()
-
-
 class Controller:
     """Caching controller for Starlette/FastAPI.
 
@@ -100,13 +72,55 @@ async def is_cachable_request(self, request: Request) -> bool:
             return False
 
         # Check Cache-Control headers
-        # todo: add parsing cache-control function
         cache_control = request.headers.get("cache-control", "").lower()
-        if "no-cache" in cache_control or "no-store" in cache_control:
+        cc = self._parse_cache_control(cache_control)
+
+        if cc.no_store or cc.no_cache or cc.private or cc.max_age == 0:
             return False
 
         return True
 
+    def _parse_cache_control(self, header: str) -> CacheControlDirectives:
+        """
+        Parse Cache-Control header into directives.
+
+        Example:
+            "max-age=60, no-cache, private"
+            ->
+            {
+                "max-age": 60,
+                "no-cache": True,
+                "private": True
+            }
+        """
+        directives: dict[str, str | int | bool | None] = {}
+
+        if not header:
+            return CacheControlDirectives()
+
+        for part in header.split(","):
+            part = part.strip()
+            if not part:
+                continue
+
+            if "=" in part:
+                key, value = part.split("=", 1)
+                key = key.lower()
+                value = value.strip().strip('"')
+
+                # numeric directives
+                if key in {"max-age", "s-maxage", "min-fresh"}:
+                    try:
+                        directives[key] = int(value)
+                    except ValueError:
+                        continue
+                else:
+                    directives[key] = value
+            else:
+                directives[part.lower()] = True
+
+        return CacheControlDirectives.model_validate(directives)
+
     async def is_cachable_response(self, response: Response) -> bool:
         """Determines if this response can be cached.
 

fast_cache_middleware/schemas.py

@@ -75,3 +75,13 @@ def path(self) -> str:
     @property
     def methods(self) -> set[str]:
         return getattr(self.route, "methods", set())
+
+
+class CacheControlDirectives(BaseModel):
+    no_cache: bool = Field(default=False, alias="no-cache")
+    no_store: bool = Field(default=False, alias="no-store")
+    private: bool = Field(default=False, alias="private")
+    max_age: int | None = Field(default=None, alias="max-age")
+    s_maxage: int | None = Field(default=None, alias="s-maxage")
+    only_if_cached: bool = Field(default=False, alias="only-if-cached")
+    no_transform: bool = Field(default=False, alias="no-transform")