From ebd3eb2481255ec2753db5d7f89483c882f62841 Mon Sep 17 00:00:00 2001 From: Nikita Yakovlev Date: Mon, 19 Jan 2026 11:51:49 +0300 Subject: [PATCH 1/6] add method for parse cache control --- fast_cache_middleware/controller.py | 44 ++++++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/fast_cache_middleware/controller.py b/fast_cache_middleware/controller.py index 068a7af..edb7bc6 100644 --- a/fast_cache_middleware/controller.py +++ b/fast_cache_middleware/controller.py @@ -2,7 +2,7 @@ import logging import re from hashlib import blake2b -from typing import Optional +from typing import Union, Optional from starlette.concurrency import run_in_threadpool from starlette.requests import Request @@ -15,6 +15,7 @@ logger = logging.getLogger(__name__) +CacheControlDirectives = dict[str, Union[bool, str, int]] KNOWN_HTTP_METHODS = [method.value for method in http.HTTPMethod] @@ -107,6 +108,47 @@ async def is_cachable_request(self, request: Request) -> bool: 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: CacheControlDirectives = {} + + if not header: + return directives + + 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 directives + async def is_cachable_response(self, response: Response) -> bool: """Determines if this response can be cached. From 91602ff2bdd6fee53c054bc898071fc7ebb80554 Mon Sep 17 00:00:00 2001 From: Nikita Yakovlev Date: Mon, 19 Jan 2026 11:53:53 +0300 Subject: [PATCH 2/6] using a new way to check cachability --- fast_cache_middleware/controller.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/fast_cache_middleware/controller.py b/fast_cache_middleware/controller.py index edb7bc6..5da3ed1 100644 --- a/fast_cache_middleware/controller.py +++ b/fast_cache_middleware/controller.py @@ -2,7 +2,7 @@ import logging import re from hashlib import blake2b -from typing import Union, Optional +from typing import Optional, Union from starlette.concurrency import run_in_threadpool from starlette.requests import Request @@ -101,9 +101,17 @@ 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 any( + [ + cc.get("no-store"), + cc.get("no-cache"), + cc.get("private"), + cc.get("max-age") == 0, + ] + ): return False return True From b3f6f432b8cbecb45fea0e2d720704c7c19afc3a Mon Sep 17 00:00:00 2001 From: Nikita Yakovlev Date: Mon, 19 Jan 2026 12:28:30 +0300 Subject: [PATCH 3/6] move generate key methods to helpers module --- fast_cache_middleware/_helpers.py | 31 +++++++++++++++++++++++++++++ fast_cache_middleware/controller.py | 1 + 2 files changed, 32 insertions(+) diff --git a/fast_cache_middleware/_helpers.py b/fast_cache_middleware/_helpers.py index e9809c1..9cc4a27 100644 --- a/fast_cache_middleware/_helpers.py +++ b/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() diff --git a/fast_cache_middleware/controller.py b/fast_cache_middleware/controller.py index 5da3ed1..d1182a0 100644 --- a/fast_cache_middleware/controller.py +++ b/fast_cache_middleware/controller.py @@ -9,6 +9,7 @@ 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 .storages import BaseStorage From 8ec43b4f247dffc6e85a8189dfcd1bb6b86c0b9f Mon Sep 17 00:00:00 2001 From: Nikita Yakovlev Date: Mon, 19 Jan 2026 12:28:46 +0300 Subject: [PATCH 4/6] change dict to Pydantic model --- fast_cache_middleware/controller.py | 56 ++++++++--------------------- 1 file changed, 15 insertions(+), 41 deletions(-) diff --git a/fast_cache_middleware/controller.py b/fast_cache_middleware/controller.py index d1182a0..0a4b822 100644 --- a/fast_cache_middleware/controller.py +++ b/fast_cache_middleware/controller.py @@ -1,8 +1,9 @@ import http import logging import re -from hashlib import blake2b -from typing import Optional, Union +from typing import Optional + +from pydantic import BaseModel, Field from starlette.concurrency import run_in_threadpool from starlette.requests import Request @@ -16,36 +17,16 @@ logger = logging.getLogger(__name__) -CacheControlDirectives = dict[str, Union[bool, str, int]] 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 CacheControlDirectives(BaseModel): + no_cache: bool = Field(default=False, alias="no-cache") + no_store: bool = Field(default=False, alias="no-store") + private: bool = False + max_age: Optional[int] = None + s_maxage: Optional[int] = 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") class Controller: @@ -105,14 +86,7 @@ async def is_cachable_request(self, request: Request) -> bool: cache_control = request.headers.get("cache-control", "").lower() cc = self._parse_cache_control(cache_control) - if any( - [ - cc.get("no-store"), - cc.get("no-cache"), - cc.get("private"), - cc.get("max-age") == 0, - ] - ): + if cc.no_store or cc.no_cache or cc.private or cc.max_age == 0: return False return True @@ -130,10 +104,10 @@ def _parse_cache_control(self, header: str) -> CacheControlDirectives: "private": True } """ - directives: CacheControlDirectives = {} + directives = {} if not header: - return directives + return CacheControlDirectives() for part in header.split(","): part = part.strip() @@ -156,7 +130,7 @@ def _parse_cache_control(self, header: str) -> CacheControlDirectives: else: directives[part.lower()] = True - return directives + return CacheControlDirectives(**directives) async def is_cachable_response(self, response: Response) -> bool: """Determines if this response can be cached. From ead1853952eb9fe0e314bff94d5f927a2e8033bb Mon Sep 17 00:00:00 2001 From: Nikita Yakovlev Date: Mon, 19 Jan 2026 13:26:33 +0300 Subject: [PATCH 5/6] lint fix --- fast_cache_middleware/controller.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/fast_cache_middleware/controller.py b/fast_cache_middleware/controller.py index 0a4b822..9f26fdd 100644 --- a/fast_cache_middleware/controller.py +++ b/fast_cache_middleware/controller.py @@ -4,7 +4,6 @@ from typing import Optional from pydantic import BaseModel, Field - from starlette.concurrency import run_in_threadpool from starlette.requests import Request from starlette.responses import Response @@ -19,12 +18,13 @@ KNOWN_HTTP_METHODS = [method.value for method in http.HTTPMethod] + class CacheControlDirectives(BaseModel): no_cache: bool = Field(default=False, alias="no-cache") no_store: bool = Field(default=False, alias="no-store") - private: bool = False - max_age: Optional[int] = None - s_maxage: Optional[int] = Field(default=None, alias="s-maxage") + 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") @@ -104,7 +104,7 @@ def _parse_cache_control(self, header: str) -> CacheControlDirectives: "private": True } """ - directives = {} + directives: dict[str, str | int | bool | None] = {} if not header: return CacheControlDirectives() @@ -130,7 +130,7 @@ def _parse_cache_control(self, header: str) -> CacheControlDirectives: else: directives[part.lower()] = True - return CacheControlDirectives(**directives) + return CacheControlDirectives.model_validate(directives) async def is_cachable_response(self, response: Response) -> bool: """Determines if this response can be cached. From 638630cda394ff155822ed40ec3f00bbad258811 Mon Sep 17 00:00:00 2001 From: Nikita Yakovlev Date: Mon, 19 Jan 2026 13:33:16 +0300 Subject: [PATCH 6/6] move model to schemas module --- fast_cache_middleware/controller.py | 13 +------------ fast_cache_middleware/schemas.py | 10 ++++++++++ 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/fast_cache_middleware/controller.py b/fast_cache_middleware/controller.py index 9f26fdd..613de63 100644 --- a/fast_cache_middleware/controller.py +++ b/fast_cache_middleware/controller.py @@ -3,7 +3,6 @@ import re from typing import Optional -from pydantic import BaseModel, Field from starlette.concurrency import run_in_threadpool from starlette.requests import Request from starlette.responses import Response @@ -11,7 +10,7 @@ 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__) @@ -19,16 +18,6 @@ KNOWN_HTTP_METHODS = [method.value for method in http.HTTPMethod] -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") - - class Controller: """Caching controller for Starlette/FastAPI. diff --git a/fast_cache_middleware/schemas.py b/fast_cache_middleware/schemas.py index abe80b1..c8676f9 100644 --- a/fast_cache_middleware/schemas.py +++ b/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")