fastapi_api.py
1 # ============================================================================ 2 # AI-Powered HR Automation with LangGraph 3 # Complete CV Review to Candidate Evaluation System 4 # Backend Server Access via FastAPI 5 6 # Get the full source code of complete project: 7 # https://aicampusmagazines.gumroad.com/l/gscdiq 8 9 ## Developed By AICampus - Gateway for future AI research & learning 10 ## Developer: Furqan Khan 11 ## Email: furqan.cloud.dev@gmail.com 12 # ============================================================================ 13 14 """ 15 FastAPI Wrapper for AI HR Automation 16 Integrates with hr_automation.py LangGraph workflow 17 Handles Forms submissions with CV file processing 18 """ 19 20 from bson import ObjectId 21 from fastapi import FastAPI, HTTPException, UploadFile, File, Form, Request 22 from fastapi.middleware.cors import CORSMiddleware 23 from fastapi.responses import JSONResponse 24 from pydantic import BaseModel, EmailStr, Field, model_validator, ConfigDict, field_serializer, field_validator 25 from pydantic.alias_generators import to_camel 26 27 from typing import Dict, Any 28 import os 29 import sys 30 import uvicorn 31 from dotenv import load_dotenv 32 import logging 33 from datetime import datetime 34 from contextlib import asynccontextmanager 35 import tempfile 36 import shutil 37 38 from pymongo import AsyncMongoClient 39 from typing import List 40 from bs4 import BeautifulSoup 41 import re 42 from typing import Optional 43 from src.config import Config 44 from utils.ulid_helper import generate_ulid 45 46 # Add parent directory to path to import hr_automation 47 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) 48 # Load environment variables 49 load_dotenv() 50 51 # Configure logging 52 logging.basicConfig( 53 level=logging.INFO, 54 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' 55 ) 56 logger = logging.getLogger(__name__) 57 58 # ============================================================================ 59 # MONGODB SETUP 60 # ============================================================================ 61 # 1. Initialize the Async Client 62 # In 2026, AsyncMongoClient is the standard for non-blocking operations 63 MONGODB_URL = os.environ.get("MONGODB_URL", "mongodb://localhost:27017") 64 client = AsyncMongoClient(MONGODB_URL) 65 db = client.get_database("ai-hr-automation") 66 67 # ============================================================================ 68 # PYDANTIC MODELS 69 # ============================================================================ 70 71 class ProcessingResult(BaseModel): 72 """Model for processing result response""" 73 success: bool 74 message: str 75 candidate_name: str 76 candidate_email: str 77 summary: str 78 score: int 79 reasoning: str 80 cv_link: str 81 timestamp: str 82 errors: List[str] = [] 83 84 85 class HealthResponse(BaseModel): 86 """Health check response""" 87 status: str 88 timestamp: str 89 service: str 90 version: str = "1.0.0" 91 config: Dict[str, str] 92 93 class User(BaseModel): 94 id: str 95 name: str 96 email: str 97 98 model_config = ConfigDict( 99 alias_generator=to_camel, 100 populate_by_name=True, 101 from_attributes=True 102 ) 103 104 @field_serializer('id') 105 def serialize_id(self, value: Optional[str]) -> Optional[str]: 106 """Convert ObjectId to string when serializing""" 107 if value and isinstance(value, ObjectId): 108 return str(value) 109 return value 110 111 112 class HRUser(User): 113 role: Optional[str] = "hr manager" 114 115 116 class JobApplication(BaseModel): 117 title: str 118 description_html: str = Field( 119 validation_alias="descriptionHTML", # For parsing incoming data 120 serialization_alias="descriptionHTML" # For serializing outgoing data 121 ) 122 description: Optional[str] = None 123 124 model_config = ConfigDict( 125 alias_generator=to_camel, 126 populate_by_name=True, 127 from_attributes=True 128 ) 129 130 @model_validator(mode='after') 131 def strip_html_and_assign(self) -> 'JobApplication': 132 # More Aggressive Line Breaking for LLM Input Prompt 133 if self.description_html: 134 soup = BeautifulSoup(self.description_html, "html.parser") 135 # Add newlines around block elements 136 block_elements = ['p', 'div', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 137 'ul', 'ol', 'blockquote', 'pre', 'hr'] 138 for tag in soup.find_all(block_elements): 139 # Add newline before and after block elements 140 tag.insert_before('\n\n') 141 tag.insert_after('\n\n') 142 143 # Handle list items 144 for li in soup.find_all('li'): 145 li.insert_before('\nā¢ ') 146 147 # Handle line breaks 148 for br in soup.find_all('br'): 149 br.replace_with('\n') 150 151 # Handle strong/bold text (keep inline) 152 for strong in soup.find_all(['strong', 'b']): 153 strong.insert_before('') 154 strong.insert_after('') 155 156 # Extract text 157 text = soup.get_text() 158 159 # Clean up excessive whitespace 160 # Collapse multiple spaces into one 161 text = re.sub(r'[ \t]+', ' ', text) 162 # Collapse 3+ newlines into 2 163 text = re.sub(r'\n{3,}', '\n\n', text) 164 # Remove spaces at start/end of lines 165 text = '\n'.join(line.strip() for line in text.split('\n')) 166 # Remove leading/trailing whitespace 167 text = text.strip() 168 169 self.description = text 170 return self 171 172 173 class HRJobPost(BaseModel): 174 id: Optional[str] = Field( 175 default=None, 176 validation_alias="_id", # For parsing incoming data 177 serialization_alias="id" # For serializing outgoing data 178 ) 179 180 ulid: Optional[str] = Field(default_factory=generate_ulid) 181 job_application: JobApplication = Field( 182 validation_alias="jobApplication", # For parsing incoming data 183 serialization_alias="jobApplication" # For serializing outgoing data 184 ) 185 hr: HRUser 186 created_at: Optional[str] = None 187 188 model_config = ConfigDict( 189 alias_generator=to_camel, 190 populate_by_name=True, 191 from_attributes=True 192 ) 193 194 @field_serializer('id') 195 def serialize_id(self, value: Optional[str]) -> Optional[str]: 196 """Convert ObjectId to string when serializing""" 197 if value and isinstance(value, ObjectId): 198 return str(value) 199 return value 200 201 @field_validator('ulid', mode='before') 202 @classmethod 203 def generate_ulid_if_missing(cls, v): 204 """Generate ULID if not provided""" 205 if v is None or v == '': 206 return generate_ulid() 207 return v 208 209 210 class CandidateSubmittedApplication(BaseModel): 211 job_id: str = Field( 212 validation_alias="jobId", # For parsing incoming data 213 serialization_alias="jobId" # For serializing outgoing data 214 ) 215 name: str 216 email: EmailStr 217 218 # Config to handle camelCase (camelCase) automatically 219 model_config = ConfigDict( 220 alias_generator=to_camel, 221 populate_by_name=True, # Allows using either snake_case or camelCase in constructor 222 from_attributes=True # Useful if converting from ORM (database) objects 223 ) 224 225 226 # ============================================================================ 227 # LIFESPAN EVENTS 228 # ============================================================================ 229 230 @asynccontextmanager 231 async def lifespan(app: FastAPI): 232 """Lifespan events for startup and shutdown""" 233 # Startup 234 logger.info("=" * 80) 235 logger.info("š Starting AI HR Automation API") 236 logger.info("=" * 80) 237 logger.info(f"Host: {Config.HOST}:{Config.PORT}") 238 logger.info("=" * 80) 239 240 # Validate configuration 241 try: 242 Config.validate() 243 logger.info("ā Configuration validated successfully") 244 except ValueError as e: 245 logger.error(f"ā Configuration validation failed: {e}") 246 logger.warning("ā ļø API will start but may not function correctly") 247 248 yield 249 250 # Shutdown 251 logger.info("š Shutting down AI HR Automation API") 252 253 254 # ============================================================================ 255 # FASTAPI APP INITIALIZATION 256 # ============================================================================ 257 258 app = FastAPI( 259 title="AI HR Automation API", 260 description="Automated CV review and candidate evaluation system powered by LangGraph and OpenAI", 261 version="1.0.0", 262 docs_url="/docs", 263 redoc_url="/redoc", 264 lifespan=lifespan 265 ) 266 267 # Configure CORS 268 app.add_middleware( 269 CORSMiddleware, 270 allow_origins=["*"], # Configure appropriately for production 271 allow_credentials=True, 272 allow_methods=["*"], 273 allow_headers=["*"], 274 ) 275 276 277 # ============================================================================ 278 # HELPER FUNCTIONS 279 # ============================================================================ 280 281 # ============================================================================ 282 # API ENDPOINTS 283 # ============================================================================ 284 285 @app.get("/", response_model=Dict[str, str]) 286 async def root(): 287 """Root endpoint with API information""" 288 return { 289 "service": "AI HR Automation API", 290 "version": "1.0.0", 291 "documentation": "/docs", 292 "health": "/health", 293 "description": "Automated CV review and candidate evaluation using LangGraph" 294 } 295 296 297 @app.get("/health", response_model=HealthResponse) 298 async def health_check(): 299 """ 300 Health check endpoint 301 Returns system status and configuration 302 """ 303 return HealthResponse( 304 status="healthy", 305 timestamp=datetime.now().isoformat(), 306 service="AI HR Automation", 307 config={ 308 "llm_provider": Config.LLM_PROVIDER 309 } 310 ) 311 312 from fastapi.exceptions import RequestValidationError 313 314 315 @app.exception_handler(RequestValidationError) 316 async def validation_exception_handler(request: Request, exc: RequestValidationError): 317 """Debug validation errors""" 318 print("Validation Error Details:") 319 print(exc.errors()) 320 print("Body:", exc.body) 321 return JSONResponse( 322 status_code=422, 323 content={"detail": exc.errors(), "body": exc.body}, 324 ) 325 326 327 @app.get("/api/config") 328 async def get_config(): 329 """ 330 Get current configuration (non-sensitive data only) 331 """ 332 return { 333 "model_provider": Config.LLM_PROVIDER, 334 "extraction_temp": Config.EXTRACTION_TEMP, 335 "summary_temp": Config.SUMMARY_TEMP, 336 "evaluation_temp": Config.EVALUATION_TEMP 337 } 338 339 340 # ============================================================================ 341 # ERROR HANDLERS 342 # ============================================================================ 343 344 @app.exception_handler(HTTPException) 345 async def http_exception_handler(request: Request, exc: HTTPException): 346 """Handle HTTP exceptions""" 347 return JSONResponse( 348 status_code=exc.status_code, 349 content={ 350 "success": False, 351 "error": exc.detail, 352 "path": str(request.url), 353 "timestamp": datetime.now().isoformat() 354 } 355 ) 356 357 358 @app.exception_handler(Exception) 359 async def general_exception_handler(request: Request, exc: Exception): 360 """Handle general exceptions""" 361 logger.error(f"Unhandled exception: {str(exc)}", exc_info=True) 362 return JSONResponse( 363 status_code=500, 364 content={ 365 "success": False, 366 "error": "Internal server error", 367 "detail": str(exc) if Config.DEBUG else "An error occurred processing your request", 368 "timestamp": datetime.now().isoformat() 369 } 370 ) 371 372 373 # ============================================================================ 374 # MAIN 375 # ============================================================================ 376 377 def get_optimal_workers(): 378 # Detect available CPU cores (logical cores) 379 cores = os.cpu_count() or 1 # Fallback to 1 if detection fails 380 381 # Apply standard production formula 382 return (2 * cores) + 1 383 384 if __name__ == "__main__": 385 logger.info("Starting FastAPI server...") 386 logger.info(f"API Documentation: http://127.0.0.1:8000/docs") 387 388 ## Get optimal workers for production 389 # workers_count = get_optimal_workers() 390 # print(f"Starting server with {workers_count} workers...") 391 392 uvicorn.run("fastapi_api:app", 393 host="127.0.0.1", 394 port=8000, 395 reload=False, 396 log_level="info", 397 # workers=workers_count 398 )