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