File size: 4,551 Bytes
363cda9 |
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 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 |
"""
CV Upload API Client.
A client for submitting job applications with CV uploads.
"""
import os
from dataclasses import dataclass
from typing import Optional, BinaryIO
import requests
def _clean_base_url(url: str) -> str:
"""Normalize base URL to avoid issues from quoted env vars."""
cleaned = url.strip().strip("\"'")
if cleaned.endswith("/"):
cleaned = cleaned[:-1]
return cleaned
@dataclass
class SubmitResponse:
"""Response from a CV submission."""
success: bool
message: str
candidate_name: str = ""
email: str = ""
cv_file_path: str = ""
already_exists: bool = False
class CVUploadClient:
"""
Client for the CV Upload API.
Usage:
client = CVUploadClient()
# Submit application
with open("my_cv.pdf", "rb") as f:
response = client.submit(
full_name="Ada Lovelace",
email="ada@example.com",
phone="+49 123 456789",
cv_file=f,
filename="my_cv.pdf"
)
if response.success:
print(f"Application submitted: {response.message}")
elif response.already_exists:
print("You already applied!")
else:
print(f"Error: {response.message}")
"""
def __init__(self, base_url: Optional[str] = None, session_id: Optional[str] = None):
"""
Initialize the CV Upload client.
Args:
base_url: API base URL. Defaults to CV_UPLOAD_API_URL env var
or http://localhost:8080/api/v1/cv
"""
raw = base_url or os.getenv(
"CV_UPLOAD_API_URL",
"http://localhost:8080/api/v1/cv"
)
self.base_url = _clean_base_url(raw)
self.session_id = (session_id or os.getenv("SESSION_ID") or "").strip().strip("\"'")
def _headers(self) -> dict:
headers = {}
if self.session_id:
headers["X-Session-Id"] = self.session_id
return headers
def submit(
self,
full_name: str,
email: str,
cv_file: BinaryIO,
filename: str,
phone: str = "",
timeout: int = 120
) -> SubmitResponse:
"""
Submit a job application with CV.
Args:
full_name: Candidate's full name
email: Candidate's email address
cv_file: File-like object containing the CV (PDF or DOCX)
filename: Original filename of the CV
phone: Optional phone number
timeout: Request timeout in seconds
Returns:
SubmitResponse with success status and details
Raises:
requests.exceptions.RequestException: On connection errors
ValueError: On API errors
"""
files = {
"cv_file": (filename, cv_file, "application/octet-stream")
}
data = {
"full_name": full_name,
"email": email,
"phone": phone,
}
response = requests.post(
f"{self.base_url}/submit",
files=files,
data=data,
headers=self._headers(),
timeout=timeout
)
if response.status_code == 400:
error = response.json().get("detail", "Invalid request")
raise ValueError(f"Validation error: {error}")
if response.status_code == 500:
error = response.json().get("detail", "Server error")
raise ValueError(f"Server error: {error}")
if response.status_code != 200:
raise ValueError(f"Unexpected status: {response.status_code}")
data = response.json()
return SubmitResponse(
success=data["success"],
message=data["message"],
candidate_name=data.get("candidate_name", ""),
email=data.get("email", ""),
cv_file_path=data.get("cv_file_path", ""),
already_exists=data.get("already_exists", False),
)
def health(self) -> bool:
"""
Check if the API is healthy.
Returns:
True if healthy, False otherwise
"""
try:
response = requests.get(f"{self.base_url}/health", timeout=5, headers=self._headers())
return response.status_code == 200
except requests.exceptions.RequestException:
return False
|