"""Read-time and durable expiry semantics for API-key assignments.""" from __future__ import annotations from datetime import date, datetime from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import and_, or_ from app.domains.apikey.models import ApiKeyAssignment SHANGHAI = ZoneInfo("Asia/Shanghai") def expiry_today() -> date: """The calendar day used by the user-facing validity contract.""" return datetime.now(SHANGHAI).date() def effective_assignment_status( status: str, valid_until: date | None, *, today: date | None = None, ) -> str: """Project a stale active row as expired without mutating it.""" current_day = today or expiry_today() if status == "active" and valid_until is not None and valid_until < current_day: return "expired" return status def effective_status_condition(status: str, *, today: date | None = None) -> Any: """SQL predicate matching the projected status used in responses.""" current_day = today or expiry_today() overdue = and_( ApiKeyAssignment.status == "active", ApiKeyAssignment.valid_until.is_not(None), ApiKeyAssignment.valid_until < current_day, ) if status == "expired": return or_(ApiKeyAssignment.status == "expired", overdue) if status == "active": return and_( ApiKeyAssignment.status == "active", or_( ApiKeyAssignment.valid_until.is_(None), ApiKeyAssignment.valid_until >= current_day, ), ) return ApiKeyAssignment.status == status __all__ = [ "effective_assignment_status", "effective_status_condition", "expiry_today", ]