145 lines
4.8 KiB
Python
145 lines
4.8 KiB
Python
"""Cars endpoints."""
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy import func, select
|
||
|
||
from ..deps import get_persistence
|
||
from ...storage.db import PersistenceService
|
||
from ...storage.models import Car, Image
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/cars")
|
||
def list_cars(
|
||
page: int = Query(1, ge=1),
|
||
per_page: int = Query(20, ge=1, le=100),
|
||
brand: str | None = None,
|
||
model: str | None = None,
|
||
year_min: int | None = None,
|
||
year_max: int | None = None,
|
||
is_sold: bool | None = None,
|
||
persistence: PersistenceService = Depends(get_persistence),
|
||
):
|
||
"""Список автомобилей с пагинацией и фильтрами."""
|
||
with persistence.session_scope() as session:
|
||
query = select(Car)
|
||
|
||
if brand:
|
||
query = query.where(Car.brand.ilike(f"%{brand}%"))
|
||
if model:
|
||
query = query.where(Car.model.ilike(f"%{model}%"))
|
||
if year_min is not None:
|
||
query = query.where(Car.year >= year_min)
|
||
if year_max is not None:
|
||
query = query.where(Car.year <= year_max)
|
||
if is_sold is not None:
|
||
query = query.where(Car.is_sold == is_sold)
|
||
|
||
# Общее количество.
|
||
count_query = select(func.count()).select_from(query.subquery())
|
||
total = session.execute(count_query).scalar() or 0
|
||
|
||
# Пагинация.
|
||
offset = (page - 1) * per_page
|
||
cars = session.execute(
|
||
query.order_by(Car.last_seen_at.desc()).offset(offset).limit(per_page)
|
||
).scalars().all()
|
||
|
||
return {
|
||
"total": total,
|
||
"page": page,
|
||
"per_page": per_page,
|
||
"pages": (total + per_page - 1) // per_page if per_page else 0,
|
||
"items": [_car_to_dict(car) for car in cars],
|
||
}
|
||
|
||
|
||
@router.get("/cars/{car_id}")
|
||
def get_car(
|
||
car_id: int,
|
||
persistence: PersistenceService = Depends(get_persistence),
|
||
):
|
||
"""Детальная информация об автомобиле с изображениями."""
|
||
with persistence.session_scope() as session:
|
||
car = session.get(Car, car_id)
|
||
if car is None:
|
||
raise HTTPException(status_code=404, detail="Car not found")
|
||
return _car_to_dict(car, include_images=True)
|
||
|
||
|
||
@router.get("/cars/by-origin/{origin_id}")
|
||
def get_car_by_origin(
|
||
origin_id: str,
|
||
persistence: PersistenceService = Depends(get_persistence),
|
||
):
|
||
"""Поиск автомобиля по origin_id."""
|
||
with persistence.session_scope() as session:
|
||
car = session.execute(
|
||
select(Car).where(Car.origin_id == origin_id)
|
||
).scalars().first()
|
||
if car is None:
|
||
raise HTTPException(status_code=404, detail="Car not found")
|
||
return _car_to_dict(car, include_images=True)
|
||
|
||
|
||
@router.get("/stats")
|
||
def get_stats(persistence: PersistenceService = Depends(get_persistence)):
|
||
"""Общая статистика по БД."""
|
||
with persistence.session_scope() as session:
|
||
total_cars = session.execute(select(func.count(Car.id))).scalar() or 0
|
||
total_images = session.execute(select(func.count(Image.id))).scalar() or 0
|
||
brands = session.execute(
|
||
select(Car.brand, func.count(Car.id))
|
||
.group_by(Car.brand)
|
||
.order_by(func.count(Car.id).desc())
|
||
.limit(20)
|
||
).all()
|
||
|
||
return {
|
||
"total_cars": total_cars,
|
||
"total_images": total_images,
|
||
"top_brands": [{"brand": b, "count": c} for b, c in brands],
|
||
}
|
||
|
||
|
||
def _car_to_dict(car: Car, include_images: bool = False) -> dict:
|
||
"""Сериализация Car в dict."""
|
||
result = {
|
||
"id": car.id,
|
||
"parser_id": car.parser_id,
|
||
"brand": car.brand,
|
||
"model": car.model,
|
||
"year": car.year,
|
||
"price": car.price,
|
||
"currency": car.currency,
|
||
"mileage": car.mileage,
|
||
"country": car.country,
|
||
"is_sold": car.is_sold,
|
||
"color": car.color,
|
||
"drive": car.drive,
|
||
"gearbox": car.gearbox,
|
||
"steering_wheel": car.steering_wheel,
|
||
"body_type": car.body_type,
|
||
"engine_volume": car.engine_volume,
|
||
"selling_type": car.selling_type,
|
||
"origin": car.origin,
|
||
"origin_url": car.origin_url,
|
||
"origin_id": car.origin_id,
|
||
"is_damaged": car.is_damaged,
|
||
"slug": car.slug,
|
||
"last_seen_at": car.last_seen_at.isoformat() if car.last_seen_at else None,
|
||
"content_hash": car.content_hash,
|
||
}
|
||
if include_images:
|
||
result["images"] = [
|
||
{
|
||
"id": img.id,
|
||
"fullres_image": img.fullres_image,
|
||
"preview_image": img.preview_image,
|
||
"order_index": img.order_index,
|
||
}
|
||
for img in sorted(car.images, key=lambda i: i.order_index)
|
||
]
|
||
return result
|