update space

app.py

@@ -1,17 +1,27 @@
 # app.py
 
-from fastapi import FastAPI, Request
+from fastapi import FastAPI, Request, HTTPException
 from fastapi.templating import Jinja2Templates
 from fastapi.responses import HTMLResponse
-from routes.authentication import auth_router
 from fastapi.staticfiles import StaticFiles
+from routes.authentication import auth_router
+import os
 
+# Initialize FastAPI app and template directory
 app = FastAPI()
 templates = Jinja2Templates(directory="templates")
-# app.mount("/static", StaticFiles(directory="static"), name="static")
-app.include_router(auth_router, prefix="/auth", tags=["Authentication"])
 
+# Mount static files if the directory exists
+static_dir = "static"
+if os.path.isdir(static_dir):
+    app.mount("/static", StaticFiles(directory=static_dir), name="static")
+else:
+    raise HTTPException(status_code=500, detail="Static directory not found.")
+
+# Include authentication routes
+app.include_router(auth_router, prefix="/auth", tags=["Authentication"])
 
 @app.get("/", response_class=HTMLResponse)
 async def landing_page(request: Request):
-    return templates.TemplateResponse("landing.html", {"request": request})
+    """Render the landing page template."""
+    return templates.TemplateResponse("landing.html", {"request": request})

requirements.txt

@@ -1,10 +1,7 @@
 fastapi~=0.115.4
-passlib[bcrypt]
-pydantic[email]
+passlib[bcrypt]~=1.7.4
+pydantic[email]~=2.9.2
 pymongo~=4.9.2
-pyjwt~=2.9.0
 python-dotenv~=1.0.1
-uvicorn~=0.32.0
-python-jose[cryptography]
-jinja2
-python-multipart
+python-jose[cryptography]~=3.3.0
+starlette~=0.41.2

routes/authentication.py

@@ -1,10 +1,12 @@
 # routes/authentication.py
 
 from datetime import timedelta, datetime
-from fastapi import APIRouter, Depends, status, Request
+from fastapi import APIRouter, Depends, status, Request, HTTPException
 from fastapi.responses import RedirectResponse, JSONResponse
 from fastapi.security import OAuth2PasswordRequestForm
 from fastapi.templating import Jinja2Templates
+from starlette.responses import HTMLResponse
+
 from models.schemas.user_schema import UserCreate, UserResponse
 from services.auth_service import (
     create_user,
@@ -20,53 +22,66 @@ auth_router = APIRouter()
 templates = Jinja2Templates(directory="templates")
 
 
-
-@auth_router.get("/register", response_class=JSONResponse)
+@auth_router.get("/register", response_class=HTMLResponse)
 async def register_page(request: Request):
-    """Renders the registration form."""
+    """Render the registration form."""
     return templates.TemplateResponse("auth/register.html", {"request": request})
 
 
 @auth_router.post("/register", response_model=UserResponse)
 async def register(request: Request, user: UserCreate):
-    """Handles user registration."""
+    """Handle user registration and redirect to login upon success."""
     result = create_user(user)
     if result["status"] == "error":
-        return templates.TemplateResponse("auth/register.html", {"request": request, "error": result["message"]})
+        return templates.TemplateResponse("auth/register.html", {
+            "request": request,
+            "error": result["message"]
+        })
     return RedirectResponse(url="/auth/login", status_code=status.HTTP_302_FOUND)
 
 
-@auth_router.get("/login", response_class=JSONResponse)
+@auth_router.get("/login", response_class=HTMLResponse)
 async def login_page(request: Request):
-    """Renders the login form."""
+    """Render the login form."""
     return templates.TemplateResponse("auth/login.html", {"request": request})
 
 
 @auth_router.post("/login", response_model=Token)
 async def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
-    """Handles user login and JWT issuance."""
+    """Authenticate user, issue JWT token, and set it in a cookie."""
     user = authenticate_user(form_data.username, form_data.password)
     if not user:
-        return templates.TemplateResponse("auth/login.html",
-                                          {"request": request, "error": "Invalid username or password"})
+        return templates.TemplateResponse("auth/login.html", {
+            "request": request,
+            "error": "Invalid username or password"
+        })
 
     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     access_token = create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires)
 
     response = RedirectResponse(url="/", status_code=status.HTTP_302_FOUND)
-    response.set_cookie(key="access_token", value=f"Bearer {access_token}", httponly=True)
+    response.set_cookie(
+        key="access_token",
+        value=f"Bearer {access_token}",
+        httponly=True,
+        max_age=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
+        secure=True,
+        samesite="Lax"
+    )
     return response
 
 
 @auth_router.get("/me", response_model=UserResponse)
 async def read_users_me(current_user: str = Depends(get_current_user)):
-    """Fetches the current authenticated user information."""
+    """Retrieve the current authenticated user information."""
+    if not current_user:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
     return {"username": current_user}
 
 
 @auth_router.get("/logout")
 async def logout(request: Request, token: str = Depends(oauth2_scheme)):
-    """Logs out the user by blacklisting the token and clearing cookies."""
+    """Logout the user by blacklisting the token and clearing cookies."""
     expiration = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     blacklist_token(token, expiration)
 

services/auth_service.py

@@ -2,12 +2,12 @@
 
 import os
 from datetime import datetime, timedelta
-from typing import Optional
+from typing import Optional, Union
 from fastapi.security import OAuth2PasswordBearer
 from passlib.context import CryptContext
 from pymongo.errors import DuplicateKeyError
 from models.schemas.user_schema import UserCreate
-from fastapi import  Depends, HTTPException, status
+from fastapi import Depends, HTTPException, status
 from pydantic import BaseModel
 from services.utils import mongo_instance
 from jose import JWTError, jwt
@@ -15,45 +15,42 @@ from jose import JWTError, jwt
 # OAuth2 scheme for FastAPI
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
 
-
 # Password hashing context
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 # JWT configuration
-SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key")  # Replace with a secure key in production
+SECRET_KEY = os.getenv("SECRET_KEY", "your-secure-secret-key")  # Replace in production
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
 # MongoDB Collection
 user_collection = mongo_instance.get_collection("users")
+blacklist_collection = mongo_instance.get_collection("token_blacklist")
 
 
 class Token(BaseModel):
     access_token: str
     token_type: str
 
-blacklist_collection = mongo_instance.get_collection("token_blacklist")
 
 def blacklist_token(token: str, expiration: datetime):
-    """Add token to blacklist."""
+    """Adds a token to the blacklist with its expiration time."""
     blacklist_collection.insert_one({"token": token, "expires": expiration})
 
+
 def is_token_blacklisted(token: str) -> bool:
-    """Check if a token is in the blacklist."""
+    """Checks if a token is blacklisted."""
     entry = blacklist_collection.find_one({"token": token})
     return entry is not None
 
-# Modify decode_access_token to check blacklist
-def decode_access_token(token: str):
+
+def decode_access_token(token: str) -> Optional[str]:
+    """Decodes and validates a JWT token, checking against the blacklist."""
     if is_token_blacklisted(token):
         return None
-    """Decode and validate JWT token."""
     try:
         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-        username: str = payload.get("sub")
-        if username is None:
-            return None
-        return username
+        return payload.get("sub")
     except JWTError:
         return None
 
@@ -63,8 +60,8 @@ def hash_password(password: str) -> str:
     return pwd_context.hash(password)
 
 
-def create_user(user_data: UserCreate):
-    """Registers a new user with hashed password."""
+def create_user(user_data: UserCreate) -> dict:
+    """Registers a new user with hashed password and stores in MongoDB."""
     try:
         hashed_password = hash_password(user_data.password)
         user_document = {
@@ -78,7 +75,7 @@ def create_user(user_data: UserCreate):
     except DuplicateKeyError:
         return {"status": "error", "message": "Username or email already exists."}
     except Exception as e:
-        return {"status": "error", "message": f"Error registering user: {e}"}
+        return {"status": "error", "message": f"Error registering user: {str(e)}"}
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
@@ -86,19 +83,18 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
     return pwd_context.verify(plain_password, hashed_password)
 
 
-def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
-    """Creates a JWT access token."""
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
+    """Creates a JWT access token with an expiration time."""
     to_encode = data.copy()
-    if expires_delta:
-        expire = datetime.utcnow() + expires_delta
-    else:
-        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
     to_encode.update({"exp": expire})
     return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
 
-def get_current_user(token: str = Depends(oauth2_scheme)):
+
+def get_current_user(token: str = Depends(oauth2_scheme)) -> Union[str, None]:
+    """Retrieves the current user by decoding the access token."""
     username = decode_access_token(token)
-    if not username:
+    if username is None:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Could not validate credentials",
@@ -107,8 +103,8 @@ def get_current_user(token: str = Depends(oauth2_scheme)):
     return username
 
 
-def authenticate_user(username: str, password: str):
-    """Authenticate user credentials."""
+def authenticate_user(username: str, password: str) -> Optional[dict]:
+    """Authenticates user credentials by checking the MongoDB records."""
     user = user_collection.find_one({"username": username})
     if user and verify_password(password, user["password"]):
         return user

services/utils.py

@@ -1,6 +1,7 @@
 # services/utils.py
 
 import os
+import logging
 from dotenv import load_dotenv
 from pymongo import MongoClient
 from pymongo.errors import ConnectionFailure, OperationFailure
@@ -11,28 +12,38 @@ load_dotenv()
 # MongoDB Connection URI
 MONGO_URI = os.getenv("MONGO_URI")
 
-# Initialize MongoDB client and database connection
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
 class MongoDB:
     def __init__(self, uri: str, db_name: str):
+        """Initializes the MongoDB client and connects to the specified database."""
         self.client = MongoClient(uri)
         self.db = self.client[db_name]
+        logger.info(f"Connected to MongoDB database: {db_name}")
 
     def test_connection(self) -> str:
-        """Tests MongoDB connection by performing a basic operation."""
+        """Tests MongoDB connection by inserting and then deleting a test document."""
         try:
-            self.db.test_collection.insert_one({"status": "connection_test"})
-            return "Connection successful: Test document inserted!"
+            test_doc = {"status": "connection_test"}
+            self.db.test_collection.insert_one(test_doc)
+            self.db.test_collection.delete_many({"status": "connection_test"})  # Cleanup
+            logger.info("Connection successful: Test document inserted and removed!")
+            return "Connection successful!"
         except (ConnectionFailure, OperationFailure) as e:
+            logger.error(f"Database connection error: {e}")
             return f"Database connection error: {e}"
 
     def get_collection(self, collection_name: str):
-        """Fetches a MongoDB collection."""
+        """Fetches a MongoDB collection with the given name."""
         return self.db[collection_name]
 
 # Instantiate MongoDB with URI and database name
 mongo_instance = MongoDB(uri=MONGO_URI, db_name="poultry_management")
 
+
 def test_db_connection():
-    """Tests the database connection and prints result."""
+    """Tests the database connection and logs the result."""
     result = mongo_instance.test_connection()
     print(result)