""" In-memory data store for the Veterinary Clinic example. This module is NOT thread-safe and is intended for demos and scaffolds only. It provides minimal, process-local data stores for the five veterinary clinic entities. Each store exposes standard CRUD operations backed by a simple dictionary. This module intentionally avoids: - persistence - concurrency guarantees - transactional semantics - validation beyond what Pydantic provides This module is not part of the ``openapi_first`` library API surface. """ from datetime import datetime, timezone from models import ( Parent, ParentCreate, Vet, VetCreate, Treatment, TreatmentCreate, Pet, PetCreate, Appointment, AppointmentCreate, ) def _now(): return datetime.now(timezone.utc).isoformat() # --------------------------------------------------------------------------- # Parents # --------------------------------------------------------------------------- _parents: dict[int, Parent] = {} _parents_next_id = 1 def list_parents() -> list[Parent]: return list(_parents.values()) def get_parent(parent_id: int) -> Parent: return _parents[parent_id] def create_parent(payload: ParentCreate) -> Parent: global _parents_next_id now = _now() parent = Parent( id=_parents_next_id, **payload.model_dump(exclude={"id", "metadata"}), metadata={"createdOn": now, "updatedOn": now} if payload.metadata else None, ) _parents[_parents_next_id] = parent _parents_next_id += 1 return parent def update_parent(parent_id: int, payload: ParentCreate) -> Parent: if parent_id not in _parents: raise KeyError(parent_id) now = _now() parent = _parents[parent_id] updated = parent.model_copy( update={ **payload.model_dump(exclude={"id", "metadata"}), "metadata": {"createdOn": parent.metadata.get("createdOn", now), "updatedOn": now} if parent.metadata else None, } ) _parents[parent_id] = updated return updated def delete_parent(parent_id: int) -> None: del _parents[parent_id] # --------------------------------------------------------------------------- # Vets # --------------------------------------------------------------------------- _vets: dict[int, Vet] = {} _vets_next_id = 1 def list_vets() -> list[Vet]: return list(_vets.values()) def get_vet(vet_id: int) -> Vet: return _vets[vet_id] def create_vet(payload: VetCreate) -> Vet: global _vets_next_id now = _now() vet = Vet( id=_vets_next_id, **payload.model_dump(exclude={"id", "metadata"}), metadata={"createdOn": now, "updatedOn": now} if payload.metadata else None, ) _vets[_vets_next_id] = vet _vets_next_id += 1 return vet def update_vet(vet_id: int, payload: VetCreate) -> Vet: if vet_id not in _vets: raise KeyError(vet_id) now = _now() vet = _vets[vet_id] updated = vet.model_copy( update={ **payload.model_dump(exclude={"id", "metadata"}), "metadata": {"createdOn": vet.metadata.get("createdOn", now), "updatedOn": now} if vet.metadata else None, } ) _vets[vet_id] = updated return updated def delete_vet(vet_id: int) -> None: del _vets[vet_id] # --------------------------------------------------------------------------- # Treatments # --------------------------------------------------------------------------- _treatments: dict[int, Treatment] = {} _treatments_next_id = 1 def list_treatments() -> list[Treatment]: return list(_treatments.values()) def get_treatment(treatment_id: int) -> Treatment: return _treatments[treatment_id] def create_treatment(payload: TreatmentCreate) -> Treatment: global _treatments_next_id now = _now() treatment = Treatment( id=_treatments_next_id, **payload.model_dump(exclude={"id", "metadata"}), metadata={"createdOn": now, "updatedOn": now} if payload.metadata else None, ) _treatments[_treatments_next_id] = treatment _treatments_next_id += 1 return treatment def update_treatment(treatment_id: int, payload: TreatmentCreate) -> Treatment: if treatment_id not in _treatments: raise KeyError(treatment_id) now = _now() treatment = _treatments[treatment_id] updated = treatment.model_copy( update={ **payload.model_dump(exclude={"id", "metadata"}), "metadata": {"createdOn": treatment.metadata.get("createdOn", now), "updatedOn": now} if treatment.metadata else None, } ) _treatments[treatment_id] = updated return updated def delete_treatment(treatment_id: int) -> None: del _treatments[treatment_id] # --------------------------------------------------------------------------- # Pets # --------------------------------------------------------------------------- _pets: dict[int, Pet] = {} _pets_next_id = 1 def list_pets() -> list[Pet]: return list(_pets.values()) def get_pet(pet_id: int) -> Pet: return _pets[pet_id] def create_pet(payload: PetCreate) -> Pet: global _pets_next_id now = _now() pet = Pet( id=_pets_next_id, **payload.model_dump(exclude={"id", "metadata"}), metadata={"createdOn": now, "updatedOn": now} if payload.metadata else None, ) _pets[_pets_next_id] = pet _pets_next_id += 1 return pet def update_pet(pet_id: int, payload: PetCreate) -> Pet: if pet_id not in _pets: raise KeyError(pet_id) now = _now() pet = _pets[pet_id] updated = pet.model_copy( update={ **payload.model_dump(exclude={"id", "metadata"}), "metadata": {"createdOn": pet.metadata.get("createdOn", now), "updatedOn": now} if pet.metadata else None, } ) _pets[pet_id] = updated return updated def delete_pet(pet_id: int) -> None: del _pets[pet_id] # --------------------------------------------------------------------------- # Appointments # --------------------------------------------------------------------------- _appointments: dict[int, Appointment] = {} _appointments_next_id = 1 def list_appointments() -> list[Appointment]: return list(_appointments.values()) def get_appointment(appointment_id: int) -> Appointment: return _appointments[appointment_id] def create_appointment(payload: AppointmentCreate) -> Appointment: global _appointments_next_id now = _now() appointment = Appointment( id=_appointments_next_id, **payload.model_dump(exclude={"id", "metadata"}), metadata={"createdOn": now, "updatedOn": now} if payload.metadata else None, ) _appointments[_appointments_next_id] = appointment _appointments_next_id += 1 return appointment def update_appointment(appointment_id: int, payload: AppointmentCreate) -> Appointment: if appointment_id not in _appointments: raise KeyError(appointment_id) now = _now() appointment = _appointments[appointment_id] updated = appointment.model_copy( update={ **payload.model_dump(exclude={"id", "metadata"}), "metadata": {"createdOn": appointment.metadata.get("createdOn", now), "updatedOn": now} if appointment.metadata else None, } ) _appointments[appointment_id] = updated return updated def delete_appointment(appointment_id: int) -> None: del _appointments[appointment_id]