Files
tfm_ainventory/venv/lib/python3.12/site-packages/websockets/sync/utils.py
Daniel Bedeleanu ea49cd6e4a feat(phase1): add image storage utilities
- Create backend/services/image_storage.py with 4 core functions:
  - sanitize_filename(): remove unsafe chars, limit to 255 chars, convert to lowercase
  - get_unique_filename(): handle collisions with UUID suffix (format: {name}_{uuid8}_{variant}.jpg)
  - ensure_image_directories(): create /images/ root and category subdirs on startup
  - save_image(): save bytes to /images/{category}/{filename}, returns relative path
- Create comprehensive test suite (22 tests) covering all functionality
- Integrate ensure_image_directories() into FastAPI startup event
- Directory structure: /images/{category}/{filename}
- Collision handling: auto-suffix with UUID if filename exists
- All tests passing, pathlib.Path for safe operations
2026-04-20 21:57:26 +03:00

46 lines
1.1 KiB
Python

from __future__ import annotations
import time
__all__ = ["Deadline"]
class Deadline:
"""
Manage timeouts across multiple steps.
Args:
timeout: Time available in seconds or :obj:`None` if there is no limit.
"""
def __init__(self, timeout: float | None) -> None:
self.deadline: float | None
if timeout is None:
self.deadline = None
else:
self.deadline = time.monotonic() + timeout
def timeout(self, *, raise_if_elapsed: bool = True) -> float | None:
"""
Calculate a timeout from a deadline.
Args:
raise_if_elapsed: Whether to raise :exc:`TimeoutError`
if the deadline lapsed.
Raises:
TimeoutError: If the deadline lapsed.
Returns:
Time left in seconds or :obj:`None` if there is no limit.
"""
if self.deadline is None:
return None
timeout = self.deadline - time.monotonic()
if raise_if_elapsed and timeout <= 0:
raise TimeoutError("timed out")
return timeout