Spaces:
Sleeping
Sleeping
File size: 3,231 Bytes
958590b 2d6c944 958590b | 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 | """
Job Application Simulator - FastAPI Server
Run with: uvicorn server.main:app --reload
"""
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List, Dict, Any
import uvicorn
from .job_app_environment import JobAppEnvironment
# Create app
app = FastAPI(
title="Job Application Simulator",
description="OpenEnv environment for training agents to apply for jobs",
version="0.1.0"
)
# CORS for local development
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Global environment instance
env = JobAppEnvironment()
# ============================================================================
# REQUEST/RESPONSE MODELS
# ============================================================================
class ResetRequest(BaseModel):
profile_name: str = "software_engineer"
budget: int = 10
difficulty: str = "normal"
class ActionRequest(BaseModel):
episode_id: str
action: Dict[str, Any]
# ============================================================================
# ENDPOINTS
# ============================================================================
@app.get("/")
async def root():
"""Health check"""
return {"status": "ok", "environment": "job-application-simulator"}
@app.get("/health")
async def health():
"""Health check endpoint for HF Spaces"""
return {"status": "healthy"}
@app.post("/reset")
async def reset(request: ResetRequest):
"""Start a new episode"""
try:
state = env.reset(
profile_name=request.profile_name,
budget=request.budget,
difficulty=request.difficulty
)
return state # Already a dict, not a Pydantic model
except Exception as e:
import traceback
return {"error": str(e), "traceback": traceback.format_exc()}
@app.post("/step")
async def step(request: ActionRequest):
"""Execute an action in the episode"""
return env.step(request.episode_id, request.action)
@app.get("/state/{episode_id}")
async def get_state(episode_id: str):
"""Get current episode state"""
result = env.step(episode_id, {"type": "state"})
if "error" in result:
raise HTTPException(status_code=404, detail=f"Episode {episode_id} not found")
return result
@app.get("/jobs")
async def list_jobs():
"""List all available jobs in the database"""
from mock_data import get_jobs
return {"jobs": get_jobs()}
@app.get("/profiles")
async def list_profiles():
"""List available applicant profiles"""
from mock_data import PROFILES
return {"profiles": list(PROFILES.keys())}
# ============================================================================
# MAIN
# ============================================================================
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=7860)
parser.add_argument("--host", default="0.0.0.0")
args = parser.parse_args()
uvicorn.run(app, host=args.host, port=args.port)
|