123 lines
4.3 KiB
Python
123 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.errors import AppError
|
|
from app.db.session import get_db
|
|
from app.models import Area, Project
|
|
from app.providers.registry import fetch_provider_data, get_provider, import_provider_dataset, list_provider_capabilities
|
|
from app.schemas import ExternalFetchRequest, ExternalFetchResponse, ProviderImportRequest
|
|
from app.utils.response import envelope
|
|
|
|
router = APIRouter(prefix="/external", tags=["external"])
|
|
|
|
|
|
def _validate_area_in_project(db: Session, project_id, area_id: str | None) -> None:
|
|
if area_id is None:
|
|
return
|
|
area = db.get(Area, area_id)
|
|
if not area:
|
|
raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
|
|
if area.project_id != project_id:
|
|
raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400)
|
|
|
|
|
|
def _assert_project_exists(db: Session, project_id):
|
|
project = db.get(Project, project_id)
|
|
if not project:
|
|
raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
|
|
|
|
|
|
def _normalize_layer_input(layers: list[str] | None) -> list[str]:
|
|
return [layer.strip() for layer in (layers or []) if isinstance(layer, str) and layer.strip()]
|
|
|
|
|
|
|
|
|
|
def _provider_payload(provider_name: str) -> dict:
|
|
return get_provider(provider_name).capability.to_dict()
|
|
|
|
|
|
@router.get("/providers")
|
|
def list_external_providers() -> dict:
|
|
return envelope({
|
|
"providers": [provider.to_dict() for provider in list_provider_capabilities()],
|
|
})
|
|
|
|
|
|
@router.get("/providers/capabilities")
|
|
def get_external_provider_capabilities() -> dict:
|
|
return envelope({
|
|
"providers": [provider.to_dict() for provider in list_provider_capabilities()],
|
|
})
|
|
|
|
|
|
@router.get("/providers/{provider_name}")
|
|
def get_external_provider(provider_name: str) -> dict:
|
|
return envelope(_provider_payload(provider_name))
|
|
|
|
|
|
@router.get("/providers/{provider_name}/layers")
|
|
def get_external_provider_layers(provider_name: str) -> dict:
|
|
provider = get_provider(provider_name)
|
|
return envelope({
|
|
"provider_name": provider.provider_name,
|
|
"layers": provider.supported_layers,
|
|
})
|
|
|
|
|
|
@router.get("/providers/{provider_name}/status")
|
|
def get_external_provider_status(provider_name: str) -> dict:
|
|
provider = get_provider(provider_name)
|
|
return envelope({
|
|
"provider_name": provider.provider_name,
|
|
"configured": provider.is_configured,
|
|
"status": provider.capability.status,
|
|
"limitation_message": provider.limitation_message,
|
|
})
|
|
|
|
|
|
@router.post("/providers/{provider_name}/import")
|
|
def import_external_provider_dataset(provider_name: str, payload: ProviderImportRequest) -> dict:
|
|
result = import_provider_dataset(
|
|
provider_name=provider_name,
|
|
project_id=payload.project_id,
|
|
area_id=payload.area_id,
|
|
layers=_normalize_layer_input(payload.layers),
|
|
requested_dataset_role=payload.dataset_role,
|
|
)
|
|
return envelope(result.model_dump())
|
|
|
|
|
|
def _run_fetch(payload: ExternalFetchRequest, provider_name: str) -> ExternalFetchResponse:
|
|
area_id_str = str(payload.area_id) if payload.area_id else None
|
|
response = fetch_provider_data(
|
|
provider_name=provider_name,
|
|
project_id=str(payload.project_id),
|
|
area_id=area_id_str,
|
|
layers=_normalize_layer_input(payload.layers),
|
|
)
|
|
return ExternalFetchResponse(
|
|
provider=provider_name,
|
|
status=response.get("status", "not_configured"),
|
|
message=response.get("message", "Provider fetch executed."),
|
|
requested_layers=_normalize_layer_input(payload.layers),
|
|
project_id=payload.project_id,
|
|
area_id=payload.area_id,
|
|
)
|
|
|
|
|
|
@router.post("/osm/fetch")
|
|
def fetch_osm(payload: ExternalFetchRequest, db: Session = Depends(get_db)) -> dict:
|
|
_assert_project_exists(db, payload.project_id)
|
|
_validate_area_in_project(db, payload.project_id, payload.area_id)
|
|
return envelope(_run_fetch(payload, "osm").model_dump())
|
|
|
|
|
|
@router.post("/grb/fetch")
|
|
def fetch_grb(payload: ExternalFetchRequest, db: Session = Depends(get_db)) -> dict:
|
|
_assert_project_exists(db, payload.project_id)
|
|
_validate_area_in_project(db, payload.project_id, payload.area_id)
|
|
return envelope(_run_fetch(payload, "grb").model_dump())
|