arbret/backend/schemas.py
counterweight 5bad1e7e17
Phase 0.1: Remove backend deprecated code
- Delete routes: counter.py, sum.py
- Delete jobs.py and worker.py
- Delete tests: test_counter.py, test_jobs.py
- Update audit.py: keep only price-history endpoints
- Update models.py: remove VIEW_COUNTER, INCREMENT_COUNTER, USE_SUM permissions
- Update models.py: remove Counter, SumRecord, CounterRecord, RandomNumberOutcome models
- Update schemas.py: remove sum/counter related schemas
- Update main.py: remove deleted router imports
- Update test_permissions.py: remove tests for deprecated features
- Update test_price_history.py: remove worker-related tests
- Update conftest.py: remove mock_enqueue_job fixture
- Update auth.py: fix example in docstring
2025-12-22 18:07:14 +01:00

251 lines
5.8 KiB
Python

"""Pydantic schemas for API request/response models."""
from datetime import date, datetime, time
from typing import Generic, TypeVar
from pydantic import BaseModel, EmailStr, field_validator
from models import InviteStatus, Permission
from shared_constants import NOTE_MAX_LENGTH
class UserCredentials(BaseModel):
"""Base model for user email/password."""
email: EmailStr
password: str
UserCreate = UserCredentials
UserLogin = UserCredentials
class UserResponse(BaseModel):
"""Response model for authenticated user info."""
id: int
email: str
roles: list[str]
permissions: list[str]
class RegisterWithInvite(BaseModel):
"""Request model for registration with invite."""
email: EmailStr
password: str
invite_identifier: str
RecordT = TypeVar("RecordT", bound=BaseModel)
class PaginatedResponse(BaseModel, Generic[RecordT]):
"""Generic paginated response wrapper."""
records: list[RecordT]
total: int
page: int
per_page: int
total_pages: int
class ProfileResponse(BaseModel):
"""Response model for profile data."""
contact_email: str | None
telegram: str | None
signal: str | None
nostr_npub: str | None
godfather_email: str | None = None
class ProfileUpdate(BaseModel):
"""Request model for updating profile."""
contact_email: str | None = None
telegram: str | None = None
signal: str | None = None
nostr_npub: str | None = None
class InviteCheckResponse(BaseModel):
"""Response for invite check endpoint."""
valid: bool
status: str | None = None
error: str | None = None
class InviteCreate(BaseModel):
"""Request model for creating an invite."""
godfather_id: int
class InviteResponse(BaseModel):
"""Response model for invite data (admin view)."""
id: int
identifier: str
godfather_id: int
godfather_email: str
status: str
used_by_id: int | None
used_by_email: str | None
created_at: datetime
spent_at: datetime | None
revoked_at: datetime | None
class UserInviteResponse(BaseModel):
"""Response model for a user's invite (simpler than admin view)."""
id: int
identifier: str
status: str
used_by_email: str | None
created_at: datetime
spent_at: datetime | None
PaginatedInviteRecords = PaginatedResponse[InviteResponse]
class AdminUserResponse(BaseModel):
"""Minimal user info for admin dropdowns."""
id: int
email: str
# =============================================================================
# Availability Schemas
# =============================================================================
class TimeSlot(BaseModel):
"""A single time slot (start and end time)."""
start_time: time
end_time: time
@field_validator("start_time", "end_time")
@classmethod
def validate_15min_boundary(cls, v: time) -> time:
"""Ensure times are on 15-minute boundaries."""
if v.minute not in (0, 15, 30, 45):
raise ValueError("Time must be on 15-minute boundary (:00, :15, :30, :45)")
if v.second != 0 or v.microsecond != 0:
raise ValueError("Time must not have seconds or microseconds")
return v
class AvailabilityDay(BaseModel):
"""Availability for a single day."""
date: date
slots: list[TimeSlot]
class AvailabilityResponse(BaseModel):
"""Response model for availability query."""
days: list[AvailabilityDay]
class SetAvailabilityRequest(BaseModel):
"""Request to set availability for a specific date."""
date: date
slots: list[TimeSlot]
class CopyAvailabilityRequest(BaseModel):
"""Request to copy availability from one day to others."""
source_date: date
target_dates: list[date]
# =============================================================================
# Booking Schemas
# =============================================================================
class BookableSlot(BaseModel):
"""A bookable 15-minute slot."""
start_time: datetime
end_time: datetime
class AvailableSlotsResponse(BaseModel):
"""Response for available slots on a given date."""
date: date
slots: list[BookableSlot]
class BookingRequest(BaseModel):
"""Request to book an appointment."""
slot_start: datetime
note: str | None = None
@field_validator("note")
@classmethod
def validate_note_length(cls, v: str | None) -> str | None:
if v is not None and len(v) > NOTE_MAX_LENGTH:
raise ValueError(f"Note must be at most {NOTE_MAX_LENGTH} characters")
return v
class AppointmentResponse(BaseModel):
"""Response model for an appointment."""
id: int
user_id: int
user_email: str
slot_start: datetime
slot_end: datetime
note: str | None
status: str
created_at: datetime
cancelled_at: datetime | None
PaginatedAppointments = PaginatedResponse[AppointmentResponse]
# =============================================================================
# Price History Schemas
# =============================================================================
class PriceHistoryResponse(BaseModel):
"""Response model for a price history record."""
id: int
source: str
pair: str
price: float
timestamp: datetime
created_at: datetime
# =============================================================================
# Meta/Constants Schemas
# =============================================================================
class ConstantsResponse(BaseModel):
"""Response model for shared constants.
Note: Using actual enum types ensures OpenAPI schema includes enum values,
allowing frontend type generation to produce matching TypeScript enums.
"""
permissions: list[Permission]
roles: list[str]
invite_statuses: list[InviteStatus]