Spaces:
Sleeping
Sleeping
File size: 6,561 Bytes
2faa7f5 9e1c5c2 2faa7f5 9e1c5c2 2faa7f5 9e1c5c2 fd2ce9d 9e1c5c2 fd2ce9d 9e1c5c2 2faa7f5 aac93aa 9e1c5c2 fd2ce9d 9e1c5c2 fd2ce9d 9e1c5c2 2faa7f5 9e1c5c2 2faa7f5 9e1c5c2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | from datetime import datetime, timezone
import uuid
from sqlalchemy.sql import text
from fastapi import HTTPException
from app.models.appointment import AppointmentCreateRequest
def build_filter_query(filters: dict):
"""
Builds a dynamic SQL filter query based on the provided filter parameters.
Args:
filters (dict): A dictionary containing filter conditions.
Returns:
str: The WHERE clause to append to a query.
"""
conditions = []
for key, value in filters.items():
if value:
conditions.append(f"{key} = :{key}")
return " AND ".join(conditions)
def validate_query_result(result, error_message="Item not found"):
"""
Validates the query result and raises an HTTPException if no result is found.
Args:
result (dict): The result of the query.
error_message (str): The error message to return if the result is None.
Raises:
HTTPException: If the result is None.
"""
if not result:
raise HTTPException(status_code=404, detail=error_message)
def validate_datetime_format(datetime_str: str):
"""
Validates if the given string is in proper datetime format (ISO 8601).
Args:
datetime_str (str): The datetime string to validate.
Raises:
HTTPException: If the datetime string is invalid.
"""
try:
datetime.fromisoformat(datetime_str)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid datetime format. Use ISO 8601 format.")
def serialize_appointment(appointment):
"""
Serializes an appointment database row into a dictionary.
Args:
appointment (RowProxy): The SQLAlchemy RowProxy object.
Returns:
dict: A serialized dictionary representation of the appointment.
"""
if appointment is None:
return None
return {
"appointment_id": str(appointment["appointment_id"]),
"customer_id": str(appointment["customer_id"]),
"merchant_id": str(appointment["merchant_id"]),
"merchant_name": appointment["merchant_name"],
"city": appointment["city"],
"location_id": appointment["location_id"],
"merchant_address": {
"street": appointment["address_street"],
"area": appointment["address_area"],
"postcode": appointment["address_in_tcode"],
"state": appointment["address_state"]
},
"appointment_date": (
appointment["appointment_date"].isoformat()
if appointment["appointment_date"] else None
),
"appointment_time": (
appointment["appointment_time"].isoformat()
if hasattr(appointment["appointment_time"], "isoformat")
else str(appointment["appointment_time"])
),
"status": appointment["status"],
"notes": appointment["notes"],
"total_amount": float(appointment["total_amount"]) if appointment["total_amount"] else 0.0,
"discount_amount": float(appointment["discount_amount"]) if appointment["discount_amount"] else 0.0,
"cleared_amount": float(appointment["cleared_amount"]) if appointment["cleared_amount"] else 0.0,
"payment_mode": appointment["payment_mode"],
"payment_status": appointment["payment_status"],
"created_at": (
appointment["created_at"].isoformat()
if appointment["created_at"] else None
),
"updated_at": (
appointment["updated_at"].isoformat()
if appointment["updated_at"] else None
),
}
def validate_existing_appointment(appointment):
"""
Validates the existing appointment's status for operations.
Args:
appointment (dict): The appointment data.
Raises:
HTTPException: If the appointment is not found or cannot be modified.
"""
if not appointment:
raise HTTPException(status_code=404, detail="Appointment not found.")
if appointment["status"] == "canceled":
raise HTTPException(status_code=400, detail="Cannot modify a canceled appointment.")
def calculate_appointment_duration(services):
"""
Calculates the total duration of the appointment based on its services.
Args:
services (list): A list of services in the appointment.
Returns:
int: Total duration in minutes.
"""
return sum(service.get("duration", 0) for service in services)
def to_in_appointments_db(appointment: AppointmentCreateRequest):
return {
"appointment_id": appointment.appointment_id,
"customer_id": appointment.customer_id, # You will update later when customer is added
"merchant_id": appointment.merchant_id,
"merchant_name": appointment.merchant_name,
"city": appointment.city,
"location_id": appointment.location_id,
# Extracted from merchant_address
"address_street": appointment.merchant_address.street,
"address_area": appointment.merchant_address.area,
"address_in_tcode": appointment.merchant_address.postcode,
"address_state": appointment.merchant_address.state,
"geo_location": (
f"POINT({appointment.merchant_address.location.coordinates[0]} "
f"{appointment.merchant_address.location.coordinates[1]})"
if appointment.merchant_address.location else None
),
"appointment_date": appointment.appointment_date,
"appointment_time": appointment.appointment_time,
"status": appointment.status,
"total_amount": appointment.total_amount,
"discount_amount": appointment.discount,
"cleared_amount": appointment.cleared_amount,
"payment_status": appointment.payment_status,
"payment_mode": appointment.payment_mode,
"notes": appointment.notes,
}
def to_in_appointments_services_db(appointment_services, appointment_id):
services_data = []
for service in appointment_services:
services_data.append({
"appointment_service_id":uuid.uuid4(),
"appointment_id": appointment_id,
"service_id": service.service_id,
"service_name": service.name,
"duration_minutes": service.duration,
"unit_price": service.price,
"quantity": service.quantity,
"line_total": service.price * service.quantity,
"associate_id": service.associate_id,
"associate_name": service.associate_name,
})
return services_data |