File size: 3,293 Bytes
1ddf755 17cb949 1ddf755 17cb949 1ddf755 df718f6 1ddf755 | 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 | from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import aiosqlite, uvicorn, os
from typing import List, Optional
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware # ← Add this line
from pydantic import BaseModel
import aiosqlite
import uvicorn
import os
from pathlib import Path
from typing import List
app = FastAPI(title="Doctor-Patient API")
# ====================== ADD THIS BLOCK ======================
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ===========================================================
# Database connection
DB_PATH = str(Path(__file__).resolve().parent.parent / "core" / "daa.db")
# Pydantic Models for Response
class Doctor(BaseModel):
id: int
doctor_name: str
category: str
visiting_days: str
visiting_time: str
visiting_money: int
class Patient(BaseModel):
id: int
doctor_name: str
doctor_category: str
patient_name: str
patient_age: str
patient_num: str
visiting_date: str
visiting_day: str | None = ""
visiting_time: str | None = ""
patient_mail: str
class AllDataResponse(BaseModel):
doctors: List[Doctor]
patients: List[Patient]
async def get_db():
"""Get database connection"""
conn = await aiosqlite.connect(DB_PATH)
conn.row_factory = aiosqlite.Row # Return rows as dictionaries
return conn
@app.get("/api/all-data", response_model=AllDataResponse)
async def get_all_data():
"""Fetch all doctors and patients data"""
conn = None
try:
conn = await get_db()
# Fetch all doctors
async with conn.execute("SELECT * FROM doctors") as cursor:
doctors_rows = await cursor.fetchall()
doctors = [dict(row) for row in doctors_rows]
# Fetch all patients
async with conn.execute("SELECT * FROM patients") as cursor:
patients_rows = await cursor.fetchall()
patients = [dict(row) for row in patients_rows]
return AllDataResponse(
doctors=doctors,
patients=patients
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
finally:
if conn:
await conn.close()
# Optional: Separate endpoints
@app.get("/api/doctors", response_model=List[Doctor])
async def get_all_doctors():
"""Get all doctors only"""
conn = None
try:
conn = await get_db()
async with conn.execute("SELECT * FROM doctors") as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]
finally:
if conn:
await conn.close()
@app.get("/api/patients", response_model=List[Patient])
async def get_all_patients():
"""Get all patients only"""
conn = None
try:
conn = await get_db()
async with conn.execute("SELECT * FROM patients") as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]
finally:
if conn:
await conn.close()
if __name__ == "__main__":
uvicorn.run("dbapi:app", host="127.0.0.1", port=8000, reload=True)
|