Spaces:
Sleeping
Sleeping
| from pydantic import BaseModel, EmailStr | |
| from typing import Optional, List, Literal, Union, Any | |
| from datetime import date, time, datetime | |
| from enum import Enum | |
| # --- USER SCHEMAS --- | |
| class SignupForm(BaseModel): | |
| email: EmailStr | |
| password: str | |
| full_name: str | |
| # Patients can only register through signup, role is automatically set to 'patient' | |
| class TokenResponse(BaseModel): | |
| access_token: str | |
| token_type: str | |
| # --- DOCTOR CREATION SCHEMA (Admin use only) --- | |
| class DoctorCreate(BaseModel): | |
| license_number: str # Changed from matricule to license_number for consistency | |
| email: EmailStr | |
| password: str | |
| full_name: str | |
| specialty: str | |
| roles: List[Literal['admin', 'doctor', 'patient']] = ['doctor'] # Can have multiple roles | |
| # --- ADMIN CREATION SCHEMA (Admin use only) --- | |
| class AdminCreate(BaseModel): | |
| email: EmailStr | |
| password: str | |
| full_name: str | |
| roles: List[Literal['admin', 'doctor', 'patient']] = ['admin'] # Can have multiple roles | |
| # --- ADMIN USER MANAGEMENT --- | |
| class AdminUserUpdate(BaseModel): | |
| full_name: Optional[str] = None | |
| roles: Optional[List[Literal['admin', 'doctor', 'patient']]] = None | |
| specialty: Optional[str] = None | |
| license_number: Optional[str] = None | |
| class AdminPasswordReset(BaseModel): | |
| new_password: str | |
| # --- PROFILE UPDATE SCHEMA --- | |
| class ProfileUpdate(BaseModel): | |
| full_name: Optional[str] = None | |
| phone: Optional[str] = None | |
| address: Optional[str] = None | |
| date_of_birth: Optional[str] = None | |
| gender: Optional[str] = None | |
| specialty: Optional[str] = None | |
| license_number: Optional[str] = None | |
| # --- PASSWORD CHANGE SCHEMA --- | |
| class PasswordChange(BaseModel): | |
| current_password: str | |
| new_password: str | |
| # --- PASSWORD RESET SCHEMAS --- | |
| class PasswordResetRequest(BaseModel): | |
| email: EmailStr | |
| class PasswordResetVerify(BaseModel): | |
| email: EmailStr | |
| verification_code: str | |
| class PasswordResetConfirm(BaseModel): | |
| email: EmailStr | |
| verification_code: str | |
| new_password: str | |
| # --- CONTACT INFO --- | |
| class ContactInfo(BaseModel): | |
| email: Optional[EmailStr] = None | |
| phone: Optional[str] = None | |
| # --- PATIENT SOURCE TYPES --- | |
| class PatientSource(str, Enum): | |
| EHR = "ehr" # Patients from external EHR system | |
| DIRECT_REGISTRATION = "direct" # Patients registered directly in the system | |
| IMPORT = "import" # Patients imported from other systems | |
| MANUAL = "manual" # Manually entered patients | |
| SYNTHEA = "synthea" # Patients imported from Synthea | |
| HAPI_FHIR = "hapi_fhir" # Patients imported from HAPI FHIR Test Server | |
| # --- PATIENT STATUS --- | |
| class PatientStatus(str, Enum): | |
| ACTIVE = "active" | |
| INACTIVE = "inactive" | |
| ARCHIVED = "archived" | |
| PENDING = "pending" | |
| # --- PATIENT SCHEMA --- | |
| class PatientCreate(BaseModel): | |
| full_name: str | |
| date_of_birth: date | |
| gender: str | |
| notes: Optional[str] = "" | |
| address: Optional[str] = None | |
| national_id: Optional[str] = None | |
| blood_type: Optional[str] = None | |
| allergies: Optional[List[str]] = [] | |
| chronic_conditions: Optional[List[str]] = [] | |
| medications: Optional[List[str]] = [] | |
| emergency_contact_name: Optional[str] = None | |
| emergency_contact_phone: Optional[str] = None | |
| insurance_provider: Optional[str] = None | |
| insurance_policy_number: Optional[str] = None | |
| contact: Optional[ContactInfo] = None | |
| # New fields for patient source management | |
| source: PatientSource = PatientSource.DIRECT_REGISTRATION | |
| ehr_id: Optional[str] = None # External EHR system ID | |
| ehr_system: Optional[str] = None # Name of the EHR system | |
| status: PatientStatus = PatientStatus.ACTIVE | |
| assigned_doctor_id: Optional[str] = None # Primary doctor assignment | |
| registration_date: Optional[datetime] = None | |
| last_visit_date: Optional[datetime] = None | |
| next_appointment_date: Optional[datetime] = None | |
| class PatientUpdate(BaseModel): | |
| full_name: Optional[str] = None | |
| date_of_birth: Optional[date] = None | |
| gender: Optional[str] = None | |
| notes: Optional[str] = None | |
| address: Optional[str] = None | |
| national_id: Optional[str] = None | |
| blood_type: Optional[str] = None | |
| allergies: Optional[List[str]] = None | |
| chronic_conditions: Optional[List[str]] = None | |
| medications: Optional[List[str]] = None | |
| emergency_contact_name: Optional[str] = None | |
| emergency_contact_phone: Optional[str] = None | |
| insurance_provider: Optional[str] = None | |
| insurance_policy_number: Optional[str] = None | |
| contact: Optional[ContactInfo] = None | |
| status: Optional[PatientStatus] = None | |
| assigned_doctor_id: Optional[str] = None | |
| last_visit_date: Optional[datetime] = None | |
| next_appointment_date: Optional[datetime] = None | |
| class PatientResponse(BaseModel): | |
| id: str | |
| full_name: str | |
| date_of_birth: Optional[date] = None | |
| gender: str | |
| notes: Optional[Union[str, List[dict]]] = [] # Can be string or list of dicts | |
| address: Optional[str] = None | |
| national_id: Optional[str] = None | |
| blood_type: Optional[str] = None | |
| allergies: List[str] = [] | |
| chronic_conditions: List[str] = [] | |
| medications: Union[List[str], List[dict]] = [] # Can be list of strings or list of dicts | |
| emergency_contact_name: Optional[str] = None | |
| emergency_contact_phone: Optional[str] = None | |
| insurance_provider: Optional[str] = None | |
| insurance_policy_number: Optional[str] = None | |
| contact: Optional[ContactInfo] = None | |
| source: PatientSource | |
| ehr_id: Optional[str] = None | |
| ehr_system: Optional[str] = None | |
| status: PatientStatus | |
| assigned_doctor_id: Optional[str] = None | |
| assigned_doctor_name: Optional[str] = None | |
| registration_date: Optional[datetime] = None | |
| last_visit_date: Optional[datetime] = None | |
| next_appointment_date: Optional[datetime] = None | |
| created_at: datetime | |
| updated_at: datetime | |
| class PatientListResponse(BaseModel): | |
| patients: List[PatientResponse] | |
| total: int | |
| page: int | |
| limit: int | |
| source_filter: Optional[PatientSource] = None | |
| status_filter: Optional[PatientStatus] = None | |
| # --- APPOINTMENT STATUS ENUM --- | |
| class AppointmentStatus(str, Enum): | |
| PENDING = "pending" | |
| CONFIRMED = "confirmed" | |
| CANCELLED = "cancelled" | |
| COMPLETED = "completed" | |
| NO_SHOW = "no_show" | |
| # --- APPOINTMENT TYPE ENUM --- | |
| class AppointmentType(str, Enum): | |
| CHECKUP = "checkup" | |
| CONSULTATION = "consultation" | |
| PROCEDURE = "procedure" | |
| FOLLOW_UP = "follow_up" | |
| EMERGENCY = "emergency" | |
| ROUTINE = "routine" | |
| # --- APPOINTMENT SCHEMAS --- | |
| class AppointmentCreate(BaseModel): | |
| patient_id: str # MongoDB ObjectId as string | |
| doctor_id: str # MongoDB ObjectId as string | |
| date: str # Date as string in 'YYYY-MM-DD' format | |
| time: str # Time as string in 'HH:MM:SS' format | |
| type: AppointmentType = AppointmentType.CONSULTATION | |
| reason: Optional[str] = None | |
| notes: Optional[str] = None | |
| duration: Optional[int] = 30 # Duration in minutes | |
| status: Optional[AppointmentStatus] = AppointmentStatus.PENDING | |
| # Enhanced patient data fields for comprehensive patient record | |
| # Personal Information | |
| patient_full_name: Optional[str] = None | |
| patient_date_of_birth: Optional[str] = None # Format: 'YYYY-MM-DD' | |
| patient_gender: Optional[str] = None | |
| patient_address: Optional[str] = None | |
| patient_city: Optional[str] = None | |
| patient_state: Optional[str] = None | |
| patient_postal_code: Optional[str] = None | |
| patient_country: Optional[str] = None | |
| patient_national_id: Optional[str] = None | |
| patient_blood_type: Optional[str] = None | |
| # Medical Information | |
| patient_allergies: Optional[List[str]] = [] | |
| patient_chronic_conditions: Optional[List[str]] = [] | |
| patient_medications: Optional[List[str]] = [] | |
| # Emergency Contact | |
| patient_emergency_contact_name: Optional[str] = None | |
| patient_emergency_contact_phone: Optional[str] = None | |
| # Insurance Information | |
| patient_insurance_provider: Optional[str] = None | |
| patient_insurance_policy_number: Optional[str] = None | |
| # Additional Notes and Encounters | |
| patient_medical_notes: Optional[str] = None | |
| patient_previous_encounters: Optional[List[dict]] = [] | |
| patient_symptoms: Optional[List[str]] = [] | |
| patient_vital_signs: Optional[dict] = None # e.g., {"blood_pressure": "120/80", "temperature": "98.6"} | |
| patient_lab_results: Optional[List[dict]] = [] | |
| patient_imaging_results: Optional[List[dict]] = [] | |
| class AppointmentUpdate(BaseModel): | |
| date: Optional[str] = None # Date as string in 'YYYY-MM-DD' format | |
| time: Optional[str] = None # Time as string in 'HH:MM:SS' format | |
| type: Optional[AppointmentType] = None | |
| reason: Optional[str] = None | |
| notes: Optional[str] = None | |
| status: Optional[AppointmentStatus] = None | |
| duration: Optional[int] = None | |
| class AppointmentResponse(BaseModel): | |
| id: str | |
| patient_id: str | |
| doctor_id: str | |
| patient_name: str | |
| doctor_name: str | |
| date: date | |
| time: time | |
| type: AppointmentType | |
| status: AppointmentStatus | |
| reason: Optional[str] = None | |
| notes: Optional[str] = None | |
| duration: int | |
| created_at: datetime | |
| updated_at: datetime | |
| class AppointmentListResponse(BaseModel): | |
| appointments: List[AppointmentResponse] | |
| total: int | |
| page: int | |
| limit: int | |
| # --- DOCTOR AVAILABILITY SCHEMAS --- | |
| class DoctorAvailabilityCreate(BaseModel): | |
| doctor_id: str | |
| day_of_week: int # 0=Monday, 6=Sunday | |
| start_time: time | |
| end_time: time | |
| is_available: bool = True | |
| class DoctorAvailabilityUpdate(BaseModel): | |
| start_time: Optional[time] = None | |
| end_time: Optional[time] = None | |
| is_available: Optional[bool] = None | |
| class DoctorAvailabilityResponse(BaseModel): | |
| id: str | |
| doctor_id: str | |
| doctor_name: str | |
| day_of_week: int | |
| start_time: time | |
| end_time: time | |
| is_available: bool | |
| # --- APPOINTMENT SLOT SCHEMAS --- | |
| class AppointmentSlot(BaseModel): | |
| date: date | |
| time: time | |
| is_available: bool | |
| appointment_id: Optional[str] = None | |
| class AvailableSlotsResponse(BaseModel): | |
| doctor_id: str | |
| doctor_name: str | |
| specialty: str | |
| date: date | |
| slots: List[AppointmentSlot] | |
| # --- DOCTOR LIST RESPONSE --- | |
| class DoctorListResponse(BaseModel): | |
| id: str | |
| full_name: str | |
| specialty: str | |
| license_number: str | |
| email: str | |
| phone: Optional[str] = None | |
| # --- MESSAGING SCHEMAS --- | |
| class MessageType(str, Enum): | |
| TEXT = "text" | |
| IMAGE = "image" | |
| FILE = "file" | |
| SYSTEM = "system" | |
| class MessageStatus(str, Enum): | |
| SENT = "sent" | |
| DELIVERED = "delivered" | |
| READ = "read" | |
| FAILED = "failed" | |
| class MessageCreate(BaseModel): | |
| recipient_id: str # MongoDB ObjectId as string | |
| content: str | |
| message_type: MessageType = MessageType.TEXT | |
| attachment_url: Optional[str] = None | |
| reply_to_message_id: Optional[str] = None | |
| class MessageUpdate(BaseModel): | |
| content: Optional[str] = None | |
| is_archived: Optional[bool] = None | |
| class MessageResponse(BaseModel): | |
| id: str | |
| sender_id: str | |
| recipient_id: str | |
| sender_name: str | |
| recipient_name: str | |
| content: str | |
| message_type: MessageType | |
| attachment_url: Optional[str] = None | |
| reply_to_message_id: Optional[str] = None | |
| status: MessageStatus | |
| is_archived: bool = False | |
| created_at: datetime | |
| updated_at: datetime | |
| read_at: Optional[datetime] = None | |
| class ConversationResponse(BaseModel): | |
| id: str | |
| participant_ids: List[str] | |
| participant_names: List[str] | |
| last_message: Optional[MessageResponse] = None | |
| unread_count: int = 0 | |
| created_at: datetime | |
| updated_at: datetime | |
| class ConversationListResponse(BaseModel): | |
| conversations: List[ConversationResponse] | |
| total: int | |
| page: int | |
| limit: int | |
| class MessageListResponse(BaseModel): | |
| messages: List[MessageResponse] | |
| total: int | |
| page: int | |
| limit: int | |
| conversation_id: str | |
| # --- NOTIFICATION SCHEMAS --- | |
| class NotificationType(str, Enum): | |
| MESSAGE = "message" | |
| APPOINTMENT = "appointment" | |
| SYSTEM = "system" | |
| REMINDER = "reminder" | |
| class NotificationPriority(str, Enum): | |
| LOW = "low" | |
| MEDIUM = "medium" | |
| HIGH = "high" | |
| URGENT = "urgent" | |
| class NotificationCreate(BaseModel): | |
| recipient_id: str | |
| title: str | |
| message: str | |
| notification_type: NotificationType | |
| priority: NotificationPriority = NotificationPriority.MEDIUM | |
| data: Optional[dict] = None # Additional data for the notification | |
| class NotificationResponse(BaseModel): | |
| id: str | |
| recipient_id: str | |
| recipient_name: str | |
| title: str | |
| message: str | |
| notification_type: NotificationType | |
| priority: NotificationPriority | |
| data: Optional[dict] = None | |
| is_read: bool = False | |
| created_at: datetime | |
| read_at: Optional[datetime] = None | |
| class NotificationListResponse(BaseModel): | |
| notifications: List[NotificationResponse] | |
| total: int | |
| unread_count: int | |
| page: int | |
| limit: int |