linting

Author: mmiqball

backend/app/middleware/auth.py

@@ -14,36 +14,39 @@
 def has_roles(required_roles: List[str]):
     """
     FastAPI dependency that checks if the authenticated user has one of the required roles.
-    
+
     Args:
         required_roles: List of roles that can access the endpoint
-        
+
     Returns:
         A dependency that validates the user has one of the specified roles
-    
+
     Example:
         @app.get("/admin-only")
         async def admin_endpoint(authorized: bool = has_roles(["admin"])):
             return {"message": "You have admin access"}
     """
+
     async def role_validator(
         request: Request,
         credentials: HTTPAuthorizationCredentials = Depends(security),
-        auth_service = Depends(get_auth_service)
+        auth_service=Depends(get_auth_service),
     ) -> bool:
         # Get the token from authorization header
         token = credentials.credentials
-        
+
         # Use the auth service to check if the user has the required role
         is_authorized = auth_service.is_authorized_by_role(token, set(required_roles))
-        
+
         if not is_authorized:
-            logger.warning(f"Access denied: user doesn't have required roles: {required_roles}")
+            logger.warning(
+                f"Access denied: user doesn't have required roles: {required_roles}"
+            )
             raise HTTPException(
                 status_code=status.HTTP_403_FORBIDDEN,
-                detail=f"Access denied: requires one of these roles: {required_roles}"
+                detail=f"Access denied: requires one of these roles: {required_roles}",
             )
-        
+
         return True
-    
-    return Depends(role_validator) 
+
+    return Depends(role_validator)

backend/app/middleware/auth_middleware.py

@@ -18,71 +18,71 @@ def is_public_path(self, path: str) -> bool:
         return path in self.public_paths
 
     async def dispatch(self, request: Request, call_next):
-
         if self.is_public_path(request.url.path):
             self.logger.info(f"Skipping auth for public path: {request.url.path}")
             return await call_next(request)
 
         # Get authentication token from header
         auth_header = request.headers.get("Authorization")
         if not auth_header or not auth_header.startswith("Bearer "):
-            self.logger.warning(f"Missing or invalid auth header for {request.url.path}")
+            self.logger.warning(
+                f"Missing or invalid auth header for {request.url.path}"
+            )
             return JSONResponse(
-                status_code=401,
-                content={"detail": "Authentication required"}
+                status_code=401, content={"detail": "Authentication required"}
             )
 
         token = auth_header.split(" ")[1]
         try:
             # Verify the token with Firebase
             self.logger.info(f"Verifying token for request to {request.url.path}")
-            decoded_token = firebase_admin.auth.verify_id_token(token, check_revoked=True)
-            
+            decoded_token = firebase_admin.auth.verify_id_token(
+                token, check_revoked=True
+            )
+
             # Get Firebase user information
             firebase_user = firebase_admin.auth.get_user(decoded_token["uid"])
-            
+
             request.state.user_id = decoded_token["uid"]
             request.state.user_email = decoded_token.get("email")
             request.state.email_verified = firebase_user.email_verified
             request.state.user_claims = decoded_token.get("claims", {})
-            
+
             # Add complete user info for convenience
             request.state.user_info = {
                 "uid": decoded_token["uid"],
                 "email": decoded_token.get("email"),
                 "name": decoded_token.get("name", ""),
                 "picture": decoded_token.get("picture", ""),
-                "email_verified": firebase_user.email_verified
+                "email_verified": firebase_user.email_verified,
             }
-            
+
             response = await call_next(request)
-            
+
             if isinstance(response, Response):
                 response.headers["X-Auth-User-ID"] = decoded_token["uid"]
-            
+
             return response
-            
+
         except firebase_admin.auth.RevokedIdTokenError:
             self.logger.warning(f"Token has been revoked: {request.url.path}")
             return JSONResponse(
-                status_code=401, 
-                content={"detail": "Token has been revoked. Please reauthenticate."}
+                status_code=401,
+                content={"detail": "Token has been revoked. Please reauthenticate."},
             )
         except firebase_admin.auth.ExpiredIdTokenError:
             self.logger.warning(f"Token has expired: {request.url.path}")
             return JSONResponse(
-                status_code=401, 
-                content={"detail": "Token has expired. Please reauthenticate."}
+                status_code=401,
+                content={"detail": "Token has expired. Please reauthenticate."},
             )
         except firebase_admin.auth.InvalidIdTokenError as e:
             self.logger.warning(f"Invalid token: {request.url.path}, error: {str(e)}")
             return JSONResponse(
-                status_code=401, 
-                content={"detail": "Invalid authentication token"}
+                status_code=401, content={"detail": "Invalid authentication token"}
             )
         except Exception as e:
             self.logger.error(f"Authentication error for {request.url.path}: {str(e)}")
             return JSONResponse(
-                status_code=401, 
-                content={"detail": f"Authentication failed: {str(e)}"}
+                status_code=401, content={"detail": f"Authentication failed: {str(e)}"}
             )

backend/app/routes/auth.py

@@ -13,24 +13,25 @@
 router = APIRouter(prefix="/auth", tags=["auth"])
 security = HTTPBearer()
 
+
 @router.post("/login", response_model=AuthResponse)
 async def login(
-    credentials: LoginRequest, 
-    auth_service: AuthService = Depends(get_auth_service)
+    credentials: LoginRequest, auth_service: AuthService = Depends(get_auth_service)
 ):
     return auth_service.generate_token(credentials.email, credentials.password)
 
+
 @router.post("/logout")
 async def logout(
     request: Request,
     credentials: HTTPAuthorizationCredentials = Depends(security),
-    auth_service: AuthService = Depends(get_auth_service)
+    auth_service: AuthService = Depends(get_auth_service),
 ):
     try:
         user_id = request.state.user_id
         if not user_id:
             raise HTTPException(status_code=401, detail="Authentication required")
-            
+
         auth_service.revoke_tokens(user_id)
         return {"message": "Successfully logged out"}
     except Exception as e:
@@ -39,11 +40,9 @@ async def logout(
 
 @router.post("/refresh", response_model=Token)
 async def refresh(
-    refresh_data: RefreshRequest,
-    auth_service: AuthService = Depends(get_auth_service)
+    refresh_data: RefreshRequest, auth_service: AuthService = Depends(get_auth_service)
 ):
     try:
         return auth_service.renew_token(refresh_data.refresh_token)
     except Exception as e:
         raise HTTPException(status_code=401, detail=str(e))
-    

backend/app/routes/user.py

@@ -25,4 +25,4 @@ async def create_user(
     except HTTPException as http_ex:
         raise http_ex
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail=str(e))

backend/app/schemas/auth.py

@@ -1,27 +1,34 @@
 from pydantic import BaseModel, ConfigDict
 from .user import UserCreateResponse
 
+
 class LoginRequest(BaseModel):
     email: str
     password: str
 
+
 class Token(BaseModel):
     """
     For authentication tokens from Firebase
     Access tokens are short-lived and used to access resources
     Refresh tokens are long-lived and used to obtain new access tokens
     """
+
     access_token: str
     refresh_token: str
-    
+
     model_config = ConfigDict(from_attributes=True)
 
+
 class RefreshRequest(BaseModel):
     """Request body for token refresh"""
+
     refresh_token: str
 
+
 class AuthResponse(Token):
     """Authentication response containing tokens and user info"""
+
     user: UserCreateResponse
-    
-    model_config = ConfigDict(from_attributes=True)
+
+    model_config = ConfigDict(from_attributes=True)

backend/app/schemas/user.py

@@ -86,4 +86,4 @@ class UserCreateResponse(BaseModel):
     auth_id: str
 
     # from_attributes enables automatic mapping from SQLAlchemy model to Pydantic model
-    model_config = ConfigDict(from_attributes=True)
+    model_config = ConfigDict(from_attributes=True)

backend/app/server.py

@@ -25,9 +25,11 @@
     "/auth/login",
     "/auth/register",
     "/health",
-    "/test-middleware-public"
+    "/test-middleware-public",
+    "/email/send-test-email",
 ]
 
+
 @asynccontextmanager
 async def lifespan(_: FastAPI):
     log.info("Starting up...")
@@ -49,10 +51,7 @@ async def lifespan(_: FastAPI):
     allow_headers=["*"],
 )
 
-app.add_middleware(
-    AuthMiddleware,
-    public_paths=PUBLIC_PATHS
-)
+app.add_middleware(AuthMiddleware, public_paths=PUBLIC_PATHS)
 
 app.include_router(auth.router)
 app.include_router(user.router)
@@ -74,9 +73,9 @@ async def test_middleware(request: Request) -> Dict[str, Any]:
     """
     Test endpoint that requires authentication and shows middleware-added state.
     This will only work if you provide a valid Firebase token in the Authorization header.
-    
+
     Example: Authorization: Bearer your-firebase-token
-    
+
     The response will show all user information added by the Firebase auth middleware.
     """
     # Get all the attributes from request.state
@@ -95,7 +94,7 @@ async def test_middleware(request: Request) -> Dict[str, Any]:
         "user_claims": getattr(request.state, "user_claims", None),
         "user_info": getattr(request.state, "user_info", None),
         "request_id": getattr(request.state, "request_id", None),
-        "authorization_header": request.headers.get("Authorization", "Not provided")
+        "authorization_header": request.headers.get("Authorization", "Not provided"),
     }
 
 
@@ -132,74 +131,74 @@ async def test_middleware_public(request: Request) -> Dict[str, Any]:
 from .middleware.auth import has_roles
 from .schemas.user import UserRole
 
+
 @app.get("/test-role-admin")
 async def test_role_admin(
-    request: Request,
-    authorized: bool = has_roles([UserRole.ADMIN])
+    request: Request, authorized: bool = has_roles([UserRole.ADMIN])
 ) -> Dict[str, Any]:
     """
     Test endpoint that requires the Admin role.
-    
+
     This demonstrates role-based access control using the has_roles dependency.
     Only users with the Admin role can access this endpoint.
     """
     return {
         "message": "You have successfully accessed an admin-only endpoint",
         "user_id": request.state.user_id,
         "user_email": request.state.user_email,
-        "role": "admin"
+        "role": "admin",
     }
 
+
 @app.get("/test-role-volunteer")
 async def test_role_volunteer(
-    request: Request,
-    authorized: bool = has_roles([UserRole.VOLUNTEER])
+    request: Request, authorized: bool = has_roles([UserRole.VOLUNTEER])
 ) -> Dict[str, Any]:
     """
     Test endpoint that requires the Volunteer role.
-    
+
     This demonstrates role-based access control using the has_roles dependency.
     Only users with the Volunteer role can access this endpoint.
     """
     return {
         "message": "You have successfully accessed a volunteer-only endpoint",
         "user_id": request.state.user_id,
         "user_email": request.state.user_email,
-        "role": "volunteer"
+        "role": "volunteer",
     }
 
+
 @app.get("/test-role-participant")
 async def test_role_participant(
-    request: Request,
-    authorized: bool = has_roles([UserRole.PARTICIPANT])
+    request: Request, authorized: bool = has_roles([UserRole.PARTICIPANT])
 ) -> Dict[str, Any]:
     """
     Test endpoint that requires the Participant role.
-    
+
     This demonstrates role-based access control using the has_roles dependency.
     Only users with the Participant role can access this endpoint.
     """
     return {
         "message": "You have successfully accessed a participant-only endpoint",
         "user_id": request.state.user_id,
         "user_email": request.state.user_email,
-        "role": "participant"
+        "role": "participant",
     }
 
+
 @app.get("/test-role-multiple")
 async def test_role_multiple(
-    request: Request,
-    authorized: bool = has_roles([UserRole.ADMIN, UserRole.VOLUNTEER])
+    request: Request, authorized: bool = has_roles([UserRole.ADMIN, UserRole.VOLUNTEER])
 ) -> Dict[str, Any]:
     """
     Test endpoint that requires either Admin OR Volunteer role.
-    
+
     This demonstrates role-based access control with multiple allowed roles.
     Users with either Admin or Volunteer roles can access this endpoint.
     """
     return {
         "message": "You have successfully accessed an endpoint requiring admin OR volunteer role",
         "user_id": request.state.user_id,
         "user_email": request.state.user_email,
-        "roles_allowed": ["admin", "volunteer"]
+        "roles_allowed": ["admin", "volunteer"],
     }