daniel156161
73fdeacbfe
Build and Push Docker Container / build-and-push (push) Successful in 2m25s
- Require Python 3.15 for the server and Python 3.14 for the CLI. - Match the server requirement to quart_common's Python 3.15 features. - Replace future annotations with native deferred evaluation. - Defer CLI command dependencies and QR imports on Python 3.15. - Enable argparse suggestions for mistyped CLI commands. - Remove the obsolete tomllib import fallback. - Update my_helpers and quart_common submodule references. - Lock test dependencies and exclude them from Docker installations. - Add regression coverage for lazy imports and deferred annotations. - Document runtime requirements and standalone CLI validation. - Validate 276 tests on Python 3.15 and 18 CLI tests on Python 3.14. BREAKING CHANGE: The server now requires Python 3.15+ and the standalone CLI requires Python 3.14+.
103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
import hashlib
|
|
import time
|
|
from email.utils import parsedate_to_datetime
|
|
|
|
def build_etag(file_id: str, file_data: dict, download: bool = False) -> str:
|
|
"""Strong validator for a stored file.
|
|
|
|
The blob behind a file_id never changes (updateFile only patches metadata),
|
|
so uploaded_at plus the storage id identify the bytes. The download flag is
|
|
part of the tag because it changes Content-Disposition.
|
|
"""
|
|
parts = [
|
|
str(file_id),
|
|
str(file_data.get('uploaded_at', '')),
|
|
str(file_data.get('db_image_url', '')),
|
|
str(file_data.get('file_name', '')),
|
|
'attachment' if download else 'inline',
|
|
]
|
|
digest = hashlib.sha256('\x1f'.join(parts).encode('utf-8')).hexdigest()[:32]
|
|
return f'"{digest}"'
|
|
|
|
def _split_etags(header: str) -> list[str]:
|
|
return [value.strip() for value in header.split(',') if value.strip()]
|
|
|
|
def etag_matches(if_none_match: str | None, etag: str) -> bool:
|
|
"""RFC 9110 If-None-Match: '*' matches, weak comparison otherwise."""
|
|
if not if_none_match or not etag:
|
|
return False
|
|
|
|
header = if_none_match.strip()
|
|
if header == '*':
|
|
return True
|
|
|
|
def normalize(value: str) -> str:
|
|
return value.removeprefix('W/')
|
|
|
|
target = normalize(etag)
|
|
return any(normalize(candidate) == target for candidate in _split_etags(header))
|
|
|
|
def _parse_http_date(value: str):
|
|
try:
|
|
return parsedate_to_datetime(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
def not_modified_since(if_modified_since: str | None, last_modified_epoch: float | None) -> bool:
|
|
if not if_modified_since or last_modified_epoch is None:
|
|
return False
|
|
|
|
parsed = _parse_http_date(if_modified_since)
|
|
if parsed is None:
|
|
return False
|
|
|
|
# HTTP dates have second precision, so compare truncated values.
|
|
return int(last_modified_epoch) <= int(parsed.timestamp())
|
|
|
|
def is_fresh(
|
|
if_none_match: str | None,
|
|
if_modified_since: str | None,
|
|
etag: str,
|
|
last_modified_epoch: float | None,
|
|
) -> bool:
|
|
"""Decide a 304 from metadata alone, before any blob is fetched."""
|
|
if if_none_match is not None:
|
|
# An ETag was offered, so it alone decides and the date is ignored.
|
|
return etag_matches(if_none_match, etag)
|
|
return not_modified_since(if_modified_since, last_modified_epoch)
|
|
|
|
def remaining_seconds(expires_at, now_ms: float | None = None) -> int | None:
|
|
"""Seconds until expires_at (epoch ms), or None when the file never expires."""
|
|
if expires_at in (None, '', 0):
|
|
return None
|
|
try:
|
|
expires_ms = float(expires_at)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
if now_ms is None:
|
|
now_ms = time.time() * 1000
|
|
return int((expires_ms - now_ms) / 1000)
|
|
|
|
def _clamped_max_age(expires_at, max_age: int) -> int:
|
|
if max_age <= 0:
|
|
return 0
|
|
remaining = remaining_seconds(expires_at)
|
|
if remaining is None:
|
|
return max_age
|
|
return min(max_age, max(0, remaining))
|
|
|
|
def cache_control(expires_at, max_age: int) -> str:
|
|
"""Browser caching for immutable blobs, clamped to the remaining lifetime."""
|
|
clamped = _clamped_max_age(expires_at, max_age)
|
|
return f'public, max-age={clamped}' if clamped > 0 else 'no-cache'
|
|
|
|
def edge_cache_control(expires_at, max_age: int) -> str:
|
|
"""Separate, deliberately short TTL for Cloudflare.
|
|
|
|
Shared caches keep serving a file after it was deleted or expired at the
|
|
origin, so the edge gets a much smaller window than the browser does.
|
|
"""
|
|
clamped = _clamped_max_age(expires_at, max_age)
|
|
return f'public, max-age={clamped}' if clamped > 0 else 'no-store'
|