cps-api-tx / models /schemas.py
Ali2206's picture
device token
1424ea9
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