shashpam/enrollmentAPI

1from pydantic import BaseModel2from fastapi import FastAPI, HTTPException3from typing import List, Optional4import csv5import os6import uvicorn7from pathlib import Path8 9app = FastAPI()10 11# Define the CSV file path12CSV_FILE = "customers.csv"13 14# Ensure the CSV file exists with headers15def init_csv():16    if not os.path.exists(CSV_FILE):17        with open(CSV_FILE, 'w', newline='') as file:18            writer = csv.DictWriter(file, fieldnames=['id', 'name', 'email', 'phone', 'address'])19            writer.writeheader()20 21# Initialize CSV file on startup22init_csv()23 24class Customer(BaseModel):25    id: int26    name: str27    email: str28    phone: Optional[str] = None29    address: Optional[str] = None30 31# Create32@app.post("/customers", response_model=Customer)33def create_customer(customer: Customer):34    # Check if customer with same ID exists35    with open(CSV_FILE, 'r', newline='') as file:36        reader = csv.DictReader(file)37        for row in reader:38            if int(row['id']) == customer.id:39                raise HTTPException(status_code=400, detail="Customer ID already exists")40    41    # Append new customer42    with open(CSV_FILE, 'a', newline='') as file:43        writer = csv.DictWriter(file, fieldnames=['id', 'name', 'email', 'phone', 'address'])44        writer.writerow(customer.dict())45    return customer46 47# Read48@app.get("/customers", response_model=List[Customer])49def get_customers():50    customers = []51    with open(CSV_FILE, 'r', newline='') as file:52        reader = csv.DictReader(file)53        for row in reader:54            # Convert string values to appropriate types55            row['id'] = int(row['id'])56            customers.append(Customer(**row))57    return customers58 59# Update60@app.put("/customers/{id}", response_model=Customer)61def update_customer(id: int, customer: Customer):62    if customer.id != id:63        raise HTTPException(status_code=400, detail="Customer ID in body must match path parameter")64    65    # Read all customers66    customers = []67    with open(CSV_FILE, 'r', newline='') as file:68        reader = csv.DictReader(file)69        customers = list(reader)70    71    # Find and update customer72    found = False73    for i, row in enumerate(customers):74        if int(row['id']) == id:75            customers[i] = customer.dict()76            found = True77            break78    79    if not found:80        raise HTTPException(status_code=404, detail="Customer not found")81    82    # Write back all customers83    with open(CSV_FILE, 'w', newline='') as file:84        writer = csv.DictWriter(file, fieldnames=['id', 'name', 'email', 'phone', 'address'])85        writer.writeheader()86        writer.writerows(customers)87    88    return customer89 90# Delete91@app.delete("/customers/{id}")92def delete_customer(id: int):93    # Read all customers94    customers = []95    with open(CSV_FILE, 'r', newline='') as file:96        reader = csv.DictReader(file)97        customers = list(reader)98    99    # Find and remove customer100    found = False101    updated_customers = []102    for row in customers:103        if int(row['id']) != id:104            updated_customers.append(row)105        else:106            found = True107    108    if not found:109        raise HTTPException(status_code=404, detail="Customer not found")110    111    # Write back remaining customers112    with open(CSV_FILE, 'w', newline='') as file:113        writer = csv.DictWriter(file, fieldnames=['id', 'name', 'email', 'phone', 'address'])114        writer.writeheader()115        writer.writerows(updated_customers)116    117    return {"message": "Customer deleted successfully"}118 119 120 121if __name__=="__main__":122    uvicorn.run(app, host="0.0.0.0", port=7860)123