Files
RekomenciBackend/src/template_project/adapters/ml_api_gateway.py
T

67 lines
2.2 KiB
Python

from collections.abc import Sequence
from decimal import Decimal
from typing import cast
from httpx import AsyncClient
from template_project.application.common.data_structure import to_data_structure
from template_project.application.resume.entity import ResumeId
@to_data_structure
class SuitableVacancyDs:
vacancy_id: str
from_salary: Decimal
to_salary: Decimal
key_skills: list[str]
resume_similarity: float
@to_data_structure
class GenerateResumePredictionResponse:
salary_from: Decimal
salary_to: Decimal
recommended_skills: list[str]
class MlApiGateway:
def __init__(self, client: AsyncClient) -> None:
self._client = client
async def generate_embedding(self, text: str) -> list[float]:
response = await self._client.post("/get_embedding", json={"text": text}, timeout=100)
return cast(list[float], response.json()["embedding"])
async def generate_resume_prediction(
self,
resume_id: ResumeId,
key_skills: list[str],
suitable_vacancies: Sequence[SuitableVacancyDs],
) -> GenerateResumePredictionResponse:
response = await self._client.post(
"/predict",
json={
"resume_id": str(resume_id),
"key_skills": key_skills,
"vacancies": [
{
"vacancy_id": str(suitable_vacancy.vacancy_id),
"from_salary": str(suitable_vacancy.from_salary),
"to_salary": str(suitable_vacancy.to_salary),
"key_skills": suitable_vacancy.key_skills,
"resume_similarity": suitable_vacancy.resume_similarity,
}
for suitable_vacancy in suitable_vacancies
],
},
timeout=100,
)
response.raise_for_status()
response_json = response.json()
return GenerateResumePredictionResponse(
salary_from=Decimal(str(response_json["salary_from"])),
salary_to=Decimal(str(response_json["salary_to"])),
recommended_skills=response_json["recommended_skills"],
)