api full working after .envwith pytest ruff unit testing cicd and improvement

Author: shresthaAlex

.github/workflows/ci.yml

@@ -0,0 +1,29 @@
+name: CI/CD Pipeline
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.11'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+        pip install ruff pytest
+    - name: Run Ruff Linting
+      run: ruff check .
+    - name: Run Tests
+      run: pytest

README.md

@@ -37,32 +37,27 @@ This is  project designed to learn how to build RESTful APIs using **FastAPI**.
 - JWT-based login using real password authentication, where the doctor ID in the token is used to authorize patient data operations
 use passlib for password hashing ,PyJWT for creating and verifying JWTs
 
-- ADD pytest for allowing  simple, scalable test writing and execution ,improves code reliability and makes it easier to catch bugs early.
+1. ADD pytest for allowing  simple, scalable test writing and execution ,improves code reliability and makes it easier to catch bugs early.
 
-
-
-## Notes for Improvement
-
-
-1. Improve Code Quality with Linting & Formatting
+2. Improve Code Quality with Linting & Formatting
        * installed ruff, b used it to
          check and format the code.
 
 
 
-   2. Organize Endpoints with API Routers
+   3. Organize Endpoints with API Routers
        * What: Your app.py file will get crowded as you add more endpoints. We can
          split the patient-related and doctor-related endpoints into separate files
          (e.g., routers/patients.py, routers/auth.py) to keep the code clean and
          modular.
 
-   3. Enhance Configuration Management
+   4. Enhance Configuration Management
        * What: Currently, you load environment variables directly in app.py. We can
          create a dedicated, type-safe config.py file using Pydantic's BaseSettings.
          This validates your settings on startup (like ensuring DATABASE_URL is set)
          and provides better autocompletion.
 
-   4. Set Up a CI/CD Pipeline
+   5. Set Up a CI/CD Pipeline
         * What: This is a professional best practice. We can create a GitHub Actions
          workflow that automatically runs your tests and linter every time you push
          new code. This acts as a gatekeeper to ensure that no broken or poorly

app.py

@@ -1,19 +1,13 @@
-import os
-from dotenv import load_dotenv
-
-load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '.env'))
-
-from fastapi import FastAPI, Path, HTTPException, Query, Depends
-from fastapi.responses import JSONResponse
-from typing import Annotated, Literal, Optional
-from fastapi.middleware.cors import CORSMiddleware
 from contextlib import asynccontextmanager
-import pymongo
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
 
 from database import init_db
-from auth import create_access_token, authenticate_user, get_current_user, get_current_doctor, verify_password, get_password_hash
-from models.patient import Patient, PatientUpdate, PatientCreate
-from models.doctor import Doctor, DoctorCreate
+from routers.auth import router as auth_router
+from routers.patients import router as patients_router
+from config import Settings
+
+settings = Settings()
 
 
 @asynccontextmanager
@@ -27,35 +21,13 @@ async def lifespan(app: FastAPI):
 
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],  # allow all origins temporarily
+    allow_origins=["*"],  # allow all origins temporarily 
+    #allow_origins=["http://localhost:3000"],  # Replace with your frontend domain(s) in production
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-@app.post("/register")
-async def register_doctor(doctor: DoctorCreate):
-    existing_doctor = await Doctor.find_one(Doctor.username == doctor.username)
-    if existing_doctor:
-        raise HTTPException(status_code=400, detail="Username already registered")
-    hashed_password = get_password_hash(doctor.password)
-    new_doctor = Doctor(username=doctor.username, password=hashed_password)
-    await new_doctor.create()
-    return {"message": "Doctor registered successfully"}
-
-@app.post("/login")
-async def login_for_access_token(form_data: DoctorCreate):
-    doctor = await Doctor.find_one(Doctor.username == form_data.username)
-    if not doctor or not verify_password(form_data.password, doctor.password):
-        raise HTTPException(
-            status_code=401,
-            detail="Incorrect username or password",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-    access_token = create_access_token(data={"sub": doctor.username})
-    return {"access_token": access_token, "token_type": "bearer"}
-        
-
 @app.get("/")
 def hello():
     return {'message':'Patient Management System API'}
@@ -64,77 +36,5 @@ def hello():
 def about():
     return {'message': 'A fully functional API to manage your patient records'}
 
-@app.get('/view')
-async def view(current_doctor: str = Depends(get_current_doctor)):
-    data = await Patient.find(Patient.doctor_id == current_doctor).to_list()
-    return data
-
-@app.get('/patient/{patient_id}')
-async def view_patient(patient_id: str = Path(..., description='ID of the patient in the DB', examples=['P001']), current_doctor: str = Depends(get_current_doctor)):
-    patient = await Patient.get(patient_id)
-    if not patient or patient.doctor_id != current_doctor:
-        raise HTTPException(status_code=404, detail='Patient not found')
-    return patient
-
-@app.get('/sort')
-async def sort_patients(sort_by: str = Query(..., description='Sort on the basis of height, weight, age or _id'), order: str = Query('asc', description='sort in asc or desc order'), current_doctor: str = Depends(get_current_doctor)):
-
-    valid_fields = ['height', 'weight', 'age', '_id']
-
-    if sort_by not in valid_fields:
-        raise HTTPException(status_code=400, detail=f'Invalid field select from {valid_fields}')
-    
-    if order not in ['asc', 'desc']:
-        raise HTTPException(status_code=400, detail='Invalid order select between asc and desc')
-    
-    sort_order = pymongo.DESCENDING if order=='desc' else pymongo.ASCENDING
-
-    sorted_data = await Patient.find(Patient.doctor_id == current_doctor).sort((sort_by,sort_order)).to_list()
-
-    return sorted_data
-
-@app.post('/create')
-async def create_patient(patient_data: PatientCreate, current_doctor: str = Depends(get_current_doctor)):
-
-    # check if the patient already exists
-    existing_patient = await Patient.get(patient_data.id)
-    if existing_patient:
-        raise HTTPException(status_code=400, detail='Patient already exists')
-
-    patient = Patient(**patient_data.model_dump(), doctor_id=current_doctor)
-    # new patient add to the database
-    await patient.create()
-
-    return JSONResponse(status_code=201, content={'message':'patient created successfully'})
-
-
-@app.put('/edit/{patient_id}')
-async def update_patient(patient_id: str, patient_update: PatientUpdate, current_doctor: str = Depends(get_current_doctor)):
-
-    patient = await Patient.get(patient_id)
-
-    if not patient or patient.doctor_id != current_doctor:
-        raise HTTPException(status_code=404, detail='Patient not found')
-    
-    patient_update_dict = patient_update.model_dump(exclude_unset=True)
-
-    for key, value in patient_update_dict.items():
-        setattr(patient, key, value)
-
-    await patient.save()
-
-
-    return JSONResponse(status_code=200, content={'message':'patient updated'})
-
-@app.delete('/delete/{patient_id}')
-async def delete_patient(patient_id: str, current_doctor: str = Depends(get_current_doctor)):
-
-    patient = await Patient.get(patient_id)
-
-    if not patient or patient.doctor_id != current_doctor:
-        raise HTTPException(status_code=404, detail='Patient not found')
-    
-    await patient.delete()
-
-
-    return JSONResponse(status_code=200, content={'message':'patient deleted'})
+app.include_router(auth_router, prefix="/auth", tags=["auth"])
+app.include_router(patients_router, prefix="/patients", tags=["patients"])

auth.py

@@ -3,10 +3,10 @@
 from passlib.context import CryptContext
 import jwt
 from datetime import datetime, timedelta
-import os
-import bcrypt
+from config import Settings
 
-SECRET_KEY = os.getenv("SECRET_KEY")
+settings = Settings()
+SECRET_KEY = settings.SECRET_KEY
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 

config.py

@@ -0,0 +1,9 @@
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    DATABASE_URL: str
+    SECRET_KEY: str
+
+    class Config:
+        env_file = ".env"
+

database.py

@@ -3,9 +3,11 @@
 import motor.motor_asyncio
 from models.patient import Patient
 from models.doctor import Doctor
-import os
+from config import Settings
+
+settings = Settings()
 
 async def init_db():
-    db_url = os.getenv("DATABASE_URL")
+    db_url = settings.DATABASE_URL
     client = motor.motor_asyncio.AsyncIOMotorClient(db_url)
     await init_beanie(database=client.db_name, document_models=[Patient, Doctor])

requirements.txt

@@ -0,0 +1,27 @@
+from fastapi import APIRouter, HTTPException
+from auth import create_access_token, verify_password, get_password_hash
+from models.doctor import Doctor, DoctorCreate
+
+router = APIRouter()
+
+@router.post("/register")
+async def register_doctor(doctor: DoctorCreate):
+    existing_doctor = await Doctor.find_one(Doctor.username == doctor.username)
+    if existing_doctor:
+        raise HTTPException(status_code=400, detail="Username already registered")
+    hashed_password = get_password_hash(doctor.password)
+    new_doctor = Doctor(username=doctor.username, password=hashed_password)
+    await new_doctor.create()
+    return {"message": "Doctor registered successfully"}
+
+@router.post("/login")
+async def login_for_access_token(form_data: DoctorCreate):
+    doctor = await Doctor.find_one(Doctor.username == form_data.username)
+    if not doctor or not verify_password(form_data.password, doctor.password):
+        raise HTTPException(
+            status_code=401,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token = create_access_token(data={"sub": doctor.username})
+    return {"access_token": access_token, "token_type": "bearer"}

routers/auth.py

@@ -0,0 +1,83 @@
+from fastapi import APIRouter, Path, HTTPException, Query, Depends
+from fastapi.responses import JSONResponse
+import pymongo
+
+from auth import get_current_doctor
+from models.patient import Patient, PatientUpdate, PatientCreate
+
+router = APIRouter()
+
+@router.get("/view")
+async def view(current_doctor: str = Depends(get_current_doctor)):
+    data = await Patient.find(Patient.doctor_id == current_doctor).to_list()
+    return data
+
+@router.get("/patient/{patient_id}")
+async def view_patient(patient_id: str = Path(..., description='ID of the patient in the DB', examples=['P001']), current_doctor: str = Depends(get_current_doctor)):
+    patient = await Patient.get(patient_id)
+    if not patient or patient.doctor_id != current_doctor:
+        raise HTTPException(status_code=404, detail='Patient not found')
+    return patient
+
+@router.get("/sort")
+async def sort_patients(sort_by: str = Query(..., description='Sort on the basis of height, weight, age or _id'), order: str = Query('asc', description='sort in asc or desc order'), current_doctor: str = Depends(get_current_doctor)):
+
+    valid_fields = ['height', 'weight', 'age', '_id']
+
+    if sort_by not in valid_fields:
+        raise HTTPException(status_code=400, detail=f'Invalid field select from {valid_fields}')
+    
+    if order not in ['asc', 'desc']:
+        raise HTTPException(status_code=400, detail='Invalid order select between asc and desc')
+    
+    sort_order = pymongo.DESCENDING if order=='desc' else pymongo.ASCENDING
+
+    sorted_data = await Patient.find(Patient.doctor_id == current_doctor).sort((sort_by,sort_order)).to_list()
+
+    return sorted_data
+
+@router.post("/create")
+async def create_patient(patient_data: PatientCreate, current_doctor: str = Depends(get_current_doctor)):
+
+    # check if the patient already exists
+    existing_patient = await Patient.get(patient_data.id)
+    if existing_patient:
+        raise HTTPException(status_code=400, detail='Patient already exists')
+
+    patient = Patient(**patient_data.model_dump(), doctor_id=current_doctor)
+    # new patient add to the database
+    await patient.create()
+
+    return JSONResponse(status_code=201, content={'message':'patient created successfully'})
+
+
+@router.put("/edit/{patient_id}")
+async def update_patient(patient_id: str, patient_update: PatientUpdate, current_doctor: str = Depends(get_current_doctor)):
+
+    patient = await Patient.get(patient_id)
+
+    if not patient or patient.doctor_id != current_doctor:
+        raise HTTPException(status_code=404, detail='Patient not found')
+    
+    patient_update_dict = patient_update.model_dump(exclude_unset=True)
+
+    for key, value in patient_update_dict.items():
+        setattr(patient, key, value)
+
+    await patient.save()
+
+
+    return JSONResponse(status_code=200, content={'message':'patient updated'})
+
+@router.delete("/delete/{patient_id}")
+async def delete_patient(patient_id: str, current_doctor: str = Depends(get_current_doctor)):
+
+    patient = await Patient.get(patient_id)
+
+    if not patient or patient.doctor_id != current_doctor:
+        raise HTTPException(status_code=404, detail='Patient not found')
+    
+    await patient.delete()
+
+
+    return JSONResponse(status_code=200, content={'message':'patient deleted'})