/
Pepegator
/
task-api
Обзор
Документация
Войти
/
Pepegator
/
task-api
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
feature/lab2
pythonProject2/idempotency.py
65 строк
2 KB
Pol136
create docs
30 дек 2025, 11:01
30 дек 2025, 11:01
d230838
Код
Авторство
О чём код?
from typing import Optional, Dict, Any from fastapi import Header, HTTPException from datetime import datetime, timedelta import hashlib import json idempotency_cache: Dict[str, Dict[str, Any]] = {} IDEMPOTENCY_TTL = timedelta(hours=24) class IdempotencyEntry: def __init__(self, response: dict, status_code: int): self.response = response self.status_code = status_code self.created_at = datetime.utcnow() def is_expired(self) -> bool: return datetime.utcnow() - self.created_at > IDEMPOTENCY_TTL def get_idempotency_key( idempotency_key: Optional[str] = Header(None) ) -> Optional[str]: if not idempotency_key: return None if len(idempotency_key) < 1 or len(idempotency_key) > 255: raise HTTPException( status_code=400, detail="Idempotency-Key должен быть от 1 до 255 символов" ) return idempotency_key def get_cached_response(idempotency_key: str) -> Optional[tuple]: if idempotency_key not in idempotency_cache: return None entry = idempotency_cache[idempotency_key] if entry.is_expired(): del idempotency_cache[idempotency_key] return None return (entry.response, entry.status_code) def cache_response( idempotency_key: str, response: dict, status_code: int ) -> None: idempotency_cache[idempotency_key] = IdempotencyEntry(response, status_code) def clear_expired_cache() -> None: expired_keys = [ key for key, entry in idempotency_cache.items() if entry.is_expired() ] for key in expired_keys: del idempotency_cache[key]