qa-and-rag-ai-assistant/routers/vacancies.py
2026-01-13 15:38:27 +03:00

70 lines
1.9 KiB
Python

from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime
from database.database import get_db, Vacancy as DBVacancy
router = APIRouter(prefix="/api/vacancies", tags=["vacancies"])
class VacancyListItem(BaseModel):
id: int
position: str
salary_min: Optional[int]
salary_max: Optional[int]
salary_currency: Optional[str]
work_arrangement: Optional[str]
key_competencies: Optional[str]
class Config:
from_attributes = True
class ContactItem(BaseModel):
id: Optional[int]
type: str
value: str
class VacancyDetail(BaseModel):
id: int
company_name: str
contacts: Optional[List[ContactItem]]
country_name: Optional[str]
created_at: datetime
description: Optional[str]
key_competencies: Optional[str]
locations: Optional[str]
position: str
relocation_supported: Optional[bool]
required_language_codes: Optional[List[str]]
salary_currency: Optional[str]
salary_min: Optional[int]
salary_max: Optional[int]
seniority_level: Optional[str]
work_arrangement: Optional[str]
class Config:
from_attributes = True
@router.get("", response_model=List[VacancyListItem])
async def get_all_vacancies(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(DBVacancy))
vacancies = result.scalars().all()
return vacancies
@router.get("/{vacancy_id}", response_model=VacancyDetail)
async def get_vacancy_by_id(vacancy_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(DBVacancy).where(DBVacancy.id == vacancy_id))
vacancy = result.scalar_one_or_none()
if not vacancy:
raise HTTPException(status_code=404, detail="Vacancy not found")
return vacancy