bookmyservice-ams / app /models /appointment.py
sivarajbookmyservice's picture
appointment new changes based on new table format
9e1c5c2
from decimal import Decimal
from sqlalchemy.dialects.postgresql import UUID as pgUUID
import uuid
from geoalchemy2 import Geography
from sqlalchemy import func
from pydantic import BaseModel, Field
from datetime import datetime, date, time
from typing import List, Dict, Optional, Any
import enum
import sqlalchemy
from app.core.sql_config import metadata
# Enum for appointment statuses
class AppointmentStatus(str, enum.Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
RESCHEDULED = "rescheduled"
COMPLETED = "completed"
CANCELED = "canceled"
# Enum for payment status
class PaymentStatus(str, enum.Enum):
PAID = "paid"
PENDING = "pending"
FAILED = "failed"
# Enum for payment mode
class PaymentMode(str, enum.Enum):
ONLINE = "online" # For Razorpay integration
OFFLINE = "offline" # For direct payments
# Model for associate assigned to the appointment
# class AssociateDetails(BaseModel):
# associate_id: str = Field(..., example="STAFF1")
# name: str = Field(..., example="John Doe")
# class Guestdetails(BaseModel):
# guest_id:str = Field(...,description="guest id")
# first_name: str =Field(..., description="first name")
# last_name: str = Field(..., description="Last name")
# class Petdetails(BaseModel):
# pet_id:str = Field(...,description="pet id")
# pet_name: str = Field(..., description="pet name")
# # Model for services booked in the appointment
# class ServiceDetails(BaseModel):
# service_id: str = Field(..., example="SERV1")
# name: str = Field(..., example="Hair Cut")
# price: float = Field(..., ge=0, example=500.00)
# duration: str = Field(..., example="30 minutes")
# quantity: int = Field(..., ge=1, example=1) # Quantity must be at least 1
# Model for Appointment
# class Appointment(BaseModel):
# appointment_id: Optional[str] = None
# merchant_id: str = Field(..., description="Merchant ID is required") # βœ… Mandatory
# merchant_name: str = Field(..., description="Merchant name is required") # βœ… Mandatory
# city: str = Field(..., description="City is required") # βœ… Mandatory
# merchant_address: Dict[str, Any] = Field(..., description="Merchant address is required") # βœ… Mandatory
# location_id: str
# customer_id: Optional[str] = Field(None, description="Customer ID is optional") # βœ… Optional
# appointment_date: Optional[date] = Field(None, description="Appointment date (YYYY-MM-DD)")
# appointment_time: str = Field(..., pattern=r"^\d{2}:\d{2}$", description="Appointment time in HH:MM format")
# associates: List[AssociateDetails] # βœ… Using `AssociateDetails` model
# guest: Optional[List[Guestdetails]] = Field(None, description="Guest details")
# pet: Optional[List[Petdetails]] = Field(None, description="Pet details")
# status: AppointmentStatus # βœ… Using Enum
# services: List[ServiceDetails] # βœ… Using `ServiceDetails` model
# notes: Optional[str] = None
# total_amount: float = Field(..., gt=0, description="Total amount must be positive") # βœ… Mandatory
# discount: float = Field(default=0.0, ge=0, description="Discount cannot be negative")
# payment_mode: PaymentMode = Field(..., description="Payment mode must be ONLINE or OFFLINE")
# payment_status: PaymentStatus = PaymentStatus.PENDING # βœ… Default to "pending"
# payment_id: Optional[str] = None
# cleared_amount: float = Field(default=0.0, ge=0, description="Cleared amount must be non-negative")
# order_id: Optional[str] = Field(None, description="Order ID is required if payment mode is ONLINE")
# business_url: Optional[str] = Field(None, description="Business URL")
# merchant_category: Optional[str] = Field(None, description="Merchant category")
# class Config:
# json_encoders = {
# datetime: lambda v: v.strftime('%Y-%m-%d %H:%M:%S'),
# date: lambda v: v.strftime('%Y-%m-%d')
# }
# json_schema_extra = {
# "example": {
# "merchant_id": "MERCHANT123",
# "merchant_name": "Elegant Salon",
# "city": "Mumbai",
# "merchant_address": {
# "line1": "12 High Street",
# "area": "Andheri East",
# "city": "Mumbai",
# "state": "MH",
# "zip_code": "400093"
# },
# "location_id": "LOC001",
# "customer_id": "CUST567",
# "appointment_date": "2025-02-25",
# "appointment_time": "14:30",
# "associates": [
# {
# "associate_id": "STAFF101",
# "name": "John Doe"
# }
# ],
# "guest": [
# {
# "guest_id":"FGYY01",
# "first_name":"John",
# "last_name":"Doe"
# }
# ],
# "pet":[
# {
# "pet_id":"GHUU01",
# "pet_name":"Buddy"
# }
# ],
# "status": "confirmed",
# "services": [
# {
# "service_id": "SERV001",
# "name": "Haircut",
# "price": 50.0,
# "quantity": 1,
# "duration": "30 minutes"
# }
# ],
# "notes": "Customer prefers short haircut",
# "total_amount": 50.0,
# "discount": 5.0,
# "payment_mode": "online",
# "payment_status": "paid",
# "payment_id": "pay_789xyz",
# "cleared_amount": 45.0,
# "order_id": "order_789xyz",
# "business_url":"johnson-gomez-and-fleming",
# "merchant_category": "spa"
# }
# }
class GeoCoordinates(BaseModel):
type: str = Field(..., example="Point")
coordinates: List[float] = Field(..., min_items=2, max_items=2)
class MerchantAddress(BaseModel):
street: str
postcode: str
state: str
location: GeoCoordinates
area: Optional[str] = None
class ServiceCreate(BaseModel):
service_id: str
name: str
duration: str # Example: "50 minutes"
price: Decimal
quantity: int
associate_name: str
associate_id: str
class AppointmentCreateRequest(BaseModel):
merchant_id: Optional[str] = None
customer_id: Optional[str] = None
appointment_id: Optional[str] = None
merchant_name: str
merchant_address: MerchantAddress
city: str
location_id: str
appointment_date: date
appointment_time: str # HH:MM format
services: List[ServiceCreate]
status: str
notes: Optional[str] = ""
total_amount: Decimal
discount: Decimal
cleared_amount: Decimal
payment_mode: str
payment_status: str
payment_id: Optional[str] = None
order_id: Optional[str] = None
class AppointmentModel(BaseModel):
appointment_id: str
customer_id: str
merchant_id: str
merchant_name: str
city: Optional[str] = None
location_id: Optional[str] = None
# Address Snapshot
address_street: Optional[str] = None
address_area: Optional[str] = None
address_in_tcode: Optional[str] = None
address_state: Optional[str] = None
# Geo location (GeoJSON format)
geo_location: Optional[Dict[str, Any]] = None
appointment_date: date
appointment_time: time
status: str
total_amount: Decimal
discount_amount: Optional[Decimal] = Decimal("0.00")
cleared_amount: Decimal
payment_status: Optional[str] = None
payment_mode: Optional[str] = None
notes: Optional[str] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
# NEW UPDATED SCHEMA - REPLACES `appointment_table`
in_appointments_table = sqlalchemy.Table(
"in_appointments",
metadata,
sqlalchemy.Column("appointment_id", pgUUID, primary_key=True,),
sqlalchemy.Column("customer_id", pgUUID, nullable=False),
sqlalchemy.Column("merchant_id", pgUUID, nullable=False),
sqlalchemy.Column("merchant_name", sqlalchemy.String, nullable=False),
sqlalchemy.Column("city", sqlalchemy.String, nullable=True),
sqlalchemy.Column("location_id", sqlalchemy.String, nullable=True),
# Address Snapshot Columns
sqlalchemy.Column("address_street", sqlalchemy.String, nullable=True),
sqlalchemy.Column("address_area", sqlalchemy.String, nullable=True),
sqlalchemy.Column("address_in_tcode", sqlalchemy.String, nullable=True),
sqlalchemy.Column("address_state", sqlalchemy.String, nullable=True),
# Geo Location (PostGIS Geography Point)
sqlalchemy.Column("geo_location", Geography(geometry_type="POINT", srid=4326), nullable=True),
sqlalchemy.Column("appointment_date", sqlalchemy.Date, nullable=False),
sqlalchemy.Column("appointment_time", sqlalchemy.Time, nullable=False),
sqlalchemy.Column("status", sqlalchemy.String, nullable=False),
sqlalchemy.Column("total_amount", sqlalchemy.Numeric(12, 2), nullable=False),
sqlalchemy.Column("discount_amount", sqlalchemy.Numeric(12, 2), nullable=True, default=0),
sqlalchemy.Column("cleared_amount", sqlalchemy.Numeric(12, 2), nullable=False),
sqlalchemy.Column("payment_status", sqlalchemy.String, nullable=True),
sqlalchemy.Column("payment_mode", sqlalchemy.String, nullable=True),
sqlalchemy.Column("notes", sqlalchemy.Text, nullable=True),
sqlalchemy.Column("reason", sqlalchemy.Text, nullable=True),
sqlalchemy.Column("created_at", sqlalchemy.DateTime(timezone=True),
server_default=func.now(), nullable=False),
sqlalchemy.Column("updated_at", sqlalchemy.DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(), nullable=False)
)
in_appointment_services_table = sqlalchemy.Table(
"in_appointment_services",
metadata,
sqlalchemy.Column("appointment_service_id", pgUUID, primary_key=True, default=uuid.uuid4),
sqlalchemy.Column("appointment_id", pgUUID, nullable=False),
sqlalchemy.Column("service_id", pgUUID, nullable=False),
sqlalchemy.Column("service_name", sqlalchemy.String, nullable=False),
sqlalchemy.Column("duration_minutes", sqlalchemy.String, nullable=False), # e.g., "50 minutes"
sqlalchemy.Column("unit_price", sqlalchemy.Numeric(12, 2), nullable=False),
sqlalchemy.Column("quantity", sqlalchemy.Integer, nullable=False),
sqlalchemy.Column("line_total", sqlalchemy.Numeric(12, 2), nullable=False),
sqlalchemy.Column("associate_name", sqlalchemy.String, nullable=False),
sqlalchemy.Column("associate_id", pgUUID, nullable=False),
sqlalchemy.Column("created_at", sqlalchemy.DateTime(timezone=True), server_default=func.now(), nullable=False),
)
# SQLAlchemy Table Definition for Appointments
# appointment_table = sqlalchemy.Table(
# "appointments",
# metadata,
# sqlalchemy.Column("appointment_id", sqlalchemy.String, primary_key=True),
# sqlalchemy.Column("merchant_name", sqlalchemy.String, nullable=False),
# sqlalchemy.Column("city", sqlalchemy.String, nullable=False),
# sqlalchemy.Column("merchant_address", sqlalchemy.JSON, nullable=False),
# sqlalchemy.Column("merchant_id", sqlalchemy.String, nullable=False),
# sqlalchemy.Column("location_id", sqlalchemy.String, nullable=False),
# sqlalchemy.Column("customer_id", sqlalchemy.String, nullable=False),
# sqlalchemy.Column("appointment_date", sqlalchemy.Date, nullable=False),
# sqlalchemy.Column("appointment_time", sqlalchemy.String(20), nullable=False), # HH:MM format
# sqlalchemy.Column("associates", sqlalchemy.JSON, nullable=False), # JSON array
# sqlalchemy.Column("guest", sqlalchemy.JSON, nullable=False), # JSON array
# sqlalchemy.Column("pet", sqlalchemy.JSON, nullable=False), # JSON array
# sqlalchemy.Column("status", sqlalchemy.Enum(AppointmentStatus, native_enum=False, values_callable=lambda enum: [e.value for e in enum]), nullable=False),
# sqlalchemy.Column("services", sqlalchemy.JSON, nullable=False), # JSON array
# sqlalchemy.Column("notes", sqlalchemy.Text, nullable=True),
# sqlalchemy.Column("total_amount", sqlalchemy.Numeric(10, 2), nullable=False), # Numeric with 2 decimal places
# sqlalchemy.Column("discount", sqlalchemy.Numeric(10, 2), nullable=True, default=0.0), # Default discount is 0.0
# sqlalchemy.Column("payment_mode", sqlalchemy.Enum(PaymentMode, native_enum=False, values_callable=lambda enum: [e.value for e in enum]), nullable=False),
# sqlalchemy.Column("payment_status", sqlalchemy.Enum(PaymentStatus, native_enum=False, values_callable=lambda enum: [e.value for e in enum]), default=PaymentStatus.PENDING.value, nullable=False),
# #sqlalchemy.Column("payment_status", sqlalchemy.Enum(PaymentStatus), nullable=False, default="pending"),
# sqlalchemy.Column("payment_id", sqlalchemy.String, nullable=True), # βœ… Added payment_id column
# sqlalchemy.Column("cleared_amount", sqlalchemy.Numeric(10, 2), nullable=False, default=0.0),
# sqlalchemy.Column("order_id", sqlalchemy.String, nullable=True), # βœ… Added order_id column
# #sqlalchemy.Column("created_at", sqlalchemy.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False),
# #sqlalchemy.Column("modified_at", sqlalchemy.DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc), nullable=False),
# sqlalchemy.Column("cancel_reason", sqlalchemy.String, nullable=True), # βœ… Added cancel_reason column
# sqlalchemy.Column("business_url", sqlalchemy.String, nullable=True),
# sqlalchemy.Column("merchant_category", sqlalchemy.String, nullable=True)
# )
# Response Model for Appointments
class AppointmentResponse(BaseModel):
appointment_id: str
merchant_id: str
city: str
merchant_name: str
merchant_address: Dict[str, Any]
customer_id: str
location_id: str
appointment_date: str
appointment_time: str
status: str
services: List[Dict[str, Any]]
notes: Optional[str]
total_amount: float
discount: float
payment_mode: str
payment_status: str
cleared_amount: float
# Model for Pagination Metadata
class PaginationMeta(BaseModel):
total: int
limit: int
offset: int
has_more: bool
current_page: int
total_pages: int
# Model for Appointment List Response with Pagination
class AppointmentListResponse(BaseModel):
customer_id: str
appointments: List[AppointmentResponse]
pagination: PaginationMeta