appointments

.env

@@ -0,0 +1,5 @@
+MONGO_URI=mongodb+srv://appliedcommerce:xaqtoj-gikdaw-6giQdy@cluster0.qpc5d.mongodb.net/test??ssl=true&ssl_cert_reqs=CERT_NONE
+
+CACHE_URI=redis-13158.c212.ap-south-1-1.ec2.redns.redis-cloud.com:13158
+
+CACHE_K=Mi5bLjJHt5KFlGjnx04K269Dt2w7hdup

app/app.py

@@ -0,0 +1,15 @@
+from fastapi import FastAPI
+from app.sql import connect_to_database, disconnect_from_database
+from app.views.appointment_view import router as appointment_router
+
+app = FastAPI(title="Bookings / Appointment Management")
+
+@app.on_event("startup")
+async def startup():
+    await connect_to_database()
+
+@app.on_event("shutdown")
+async def shutdown():
+    await disconnect_from_database()
+
+app.include_router(appointment_router)

app/controllers/appointment_controller.py

@@ -0,0 +1,13 @@
+from fastapi import APIRouter
+from app.services.appointment_service import create_new_appointment, reschedule_appointment
+from app.models.appointment_model import Appointment
+
+router = APIRouter()
+
+@router.post("/appointments/")
+async def create_appointment(appointment: Appointment):
+    return await create_new_appointment(appointment)
+
+@router.put("/appointments/reschedule/{appointment_id}")
+async def reschedule(appointment_id: str, new_date: str, new_time: str):
+    return await reschedule_appointment(appointment_id, new_date, new_time)

app/models/appointment_model.py

@@ -0,0 +1,41 @@
+from pydantic import BaseModel, Field
+from datetime import date, time, datetime
+from typing import List
+from enum import Enum
+import sqlalchemy
+from app.database import metadata
+
+class AppointmentStatus(str, Enum):
+    confirmed = "confirmed"
+    rescheduled = "rescheduled"
+    canceled = "canceled"
+
+class Service(BaseModel):
+    description: str
+    duration: int
+    price: float
+
+class Appointment(BaseModel):
+    appointment_id: str
+    merchant_id: str
+    customer_id: str
+    appointment_date: date
+    appointment_time: time
+    status: AppointmentStatus
+    services: List[Service]
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    modified_at: datetime = Field(default_factory=datetime.utcnow)
+
+appointment_table = sqlalchemy.Table(
+    "appointments",
+    metadata,
+    sqlalchemy.Column("appointment_id", sqlalchemy.String, primary_key=True),
+    sqlalchemy.Column("merchant_id", sqlalchemy.String),
+    sqlalchemy.Column("customer_id", sqlalchemy.String),
+    sqlalchemy.Column("appointment_date", sqlalchemy.Date, nullable=False),
+    sqlalchemy.Column("appointment_time", sqlalchemy.Time, nullable=False),
+    sqlalchemy.Column("status", sqlalchemy.String(20)),
+    sqlalchemy.Column("services", sqlalchemy.JSON),
+    sqlalchemy.Column("created_at", sqlalchemy.DateTime(timezone=True), default=datetime.utcnow),
+    sqlalchemy.Column("modified_at", sqlalchemy.DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow),
+)

app/repositories/appointment_repository.py

@@ -0,0 +1,26 @@
+from app.models.appointment_model import appointment_table
+from app.database import database
+
+async def create_appointment(appointment_data: dict):
+    query = appointment_table.insert().values(**appointment_data)
+    return await database.execute(query)
+
+async def update_appointment(appointment_id: str, update_data: dict):
+    query = (
+        appointment_table.update()
+        .where(appointment_table.c.appointment_id == appointment_id)
+        .values(**update_data)
+    )
+    return await database.execute(query)
+
+async def get_appointment_by_id(appointment_id: str):
+    query = appointment_table.select().where(appointment_table.c.appointment_id == appointment_id)
+    return await database.fetch_one(query)
+
+async def get_appointments_by_customer(customer_id: str):
+    query = appointment_table.select().where(appointment_table.c.customer_id == customer_id)
+    return await database.fetch_all(query)
+
+async def get_appointments_by_merchant(merchant_id: str):
+    query = appointment_table.select().where(appointment_table.c.merchant_id == merchant_id)
+    return await database.fetch_all(query)

app/services/appointment_service.py

@@ -0,0 +1,34 @@
+from datetime import datetime, timezone
+from fastapi import HTTPException
+from app.repositories.appointment_repository import (
+    create_appointment,
+    update_appointment,
+    get_appointment_by_id,
+)
+from app.models.appointment_model import Appointment, AppointmentStatus
+
+async def create_new_appointment(appointment: Appointment):
+    now = datetime.now(timezone.utc)
+    appointment_data = appointment.dict()
+    appointment_data["status"] = AppointmentStatus.confirmed
+    appointment_data["created_at"] = now
+    appointment_data["modified_at"] = now
+    await create_appointment(appointment_data)
+    return {"message": "Appointment created successfully"}
+
+async def reschedule_appointment(appointment_id: str, new_date: datetime, new_time: datetime):
+    existing_appointment = await get_appointment_by_id(appointment_id)
+    if not existing_appointment:
+        raise HTTPException(status_code=404, detail="Appointment not found")
+
+    if existing_appointment["status"] == AppointmentStatus.canceled:
+        raise HTTPException(status_code=400, detail="Cannot reschedule a canceled appointment")
+
+    update_data = {
+        "appointment_date": new_date.date(),
+        "appointment_time": new_time.time(),
+        "status": AppointmentStatus.rescheduled,
+        "modified_at": datetime.now(timezone.utc),
+    }
+    await update_appointment(appointment_id, update_data)
+    return {"message": "Appointment rescheduled successfully"}

app/sql.py

@@ -0,0 +1,55 @@
+import os
+import logging
+from dotenv import load_dotenv
+import databases
+import sqlalchemy
+
+# Load environment variables from .env file
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# Load the DATABASE_URL from environment variables
+DATABASE_URL = os.getenv("DATABASE_URI")
+
+if not DATABASE_URL:
+    logger.error("DATABASE_URI not found in environment variables.")
+    raise ValueError("DATABASE_URI not found in environment variables. Ensure it is set in the .env file.")
+
+# Initialize the database connection and metadata
+try:
+    database = databases.Database(DATABASE_URL)
+    metadata = sqlalchemy.MetaData()
+    logger.info("Database connection initialized successfully.")
+except Exception as e:
+    logger.error("Failed to initialize database connection: %s", e)
+    raise
+
+# Helper function to initialize database
+async def connect_to_database():
+    """
+    Connects to the database when the application starts.
+    """
+    try:
+        await database.connect()
+        logger.info("Successfully connected to the database.")
+    except Exception as e:
+        logger.error("Error connecting to the database: %s", e)
+        raise
+
+# Helper function to disconnect database
+async def disconnect_from_database():
+    """
+    Disconnects from the database when the application shuts down.
+    """
+    try:
+        await database.disconnect()
+        logger.info("Successfully disconnected from the database.")
+    except Exception as e:
+        logger.error("Error disconnecting from the database: %s", e)
+        raise

app/views/appointment_view.py

@@ -0,0 +1,5 @@
+from fastapi import APIRouter
+from app.controllers.appointment_controller import router as appointment_controller
+
+router = APIRouter()
+router.include_router(appointment_controller, prefix="/appointments", tags=["Appointments"])