auto-generate mapdata api endpoints

This commit is contained in:
Laura Klünder 2024-12-02 23:40:41 +01:00
parent aec56e5bbb
commit f70040fe1d

View file

@ -1,11 +1,15 @@
from typing import Optional, Sequence, Type
from collections import namedtuple
from dataclasses import dataclass
from typing import Optional, Sequence, Type, Callable, Any
from django.db.models import Model
from ninja import Query
from ninja import Router as APIRouter
from pydantic import PositiveInt
from c3nav.api.auth import auth_responses, validate_responses
from c3nav.api.exceptions import API404
from c3nav.api.schema import BaseSchema
from c3nav.mapdata.api.base import api_etag, optimize_query
from c3nav.mapdata.models import (Area, Building, Door, Hole, Level, LocationGroup, LocationGroupCategory, Source,
Space, Stair, DataOverlay, DataOverlayFeature)
@ -14,7 +18,7 @@ from c3nav.mapdata.models.geometry.space import (POI, Column, CrossDescription,
Obstacle, Ramp)
from c3nav.mapdata.models.locations import DynamicLocation
from c3nav.mapdata.schemas.filters import (ByCategoryFilter, ByGroupFilter, ByOnTopOfFilter, FilterSchema,
LevelGeometryFilter, SpaceGeometryFilter)
LevelGeometryFilter, SpaceGeometryFilter, BySpaceFilter)
from c3nav.mapdata.schemas.model_base import schema_description
from c3nav.mapdata.schemas.models import (AccessRestrictionGroupSchema, AccessRestrictionSchema, AreaSchema,
BuildingSchema, ColumnSchema, CrossDescriptionSchema, DoorSchema,
@ -58,436 +62,214 @@ def mapdata_retrieve_endpoint(request, model: Type[Model], **lookups):
raise API404("%s not found" % model.__name__.lower())
"""
Levels
"""
@dataclass
class MapdataEndpoint:
model: Type[Model]
schema: Type[BaseSchema]
filters: Type[FilterSchema] | None = None
@property
def model_name(self):
return self.model._meta.model_name
@property
def model_name_plural(self):
return self.model._meta.default_related_name
@dataclass
class MapdataAPIBuilder:
router: APIRouter
def build_all_endpoints(self, endpoints: dict[str, list[MapdataEndpoint]]):
for tag, endpoints in endpoints.items():
for endpoint in endpoints:
self.add_endpoints(endpoint, tag=tag)
def add_endpoints(self, endpoint: MapdataEndpoint, tag: str):
self.add_list_endpoint(endpoint, tag=tag)
self.add_by_id_endpoint(endpoint, tag=tag)
def common_params(self, endpoint: MapdataEndpoint) -> dict[str: Any]:
return {"request": None}
def _make_endpoint(self, view_params: dict[str, str], call_func: Callable,
add_call_params: dict[str, str] = None) -> Callable:
if add_call_params is None:
add_call_params = {}
call_param_values = set(add_call_params.values())
call_params = (
*(f"{name}={name}" for name in set(view_params.keys())-call_param_values),
*(f"{name}={value}" for name, value in add_call_params.items()),
)
method_code = "\n".join((
f"def gen_func({", ".join((f"{name}: {hint}" if hint else name) for name, hint in view_params.items())}):",
f" call_func({", ".join(call_params)})",
))
exec(method_code, globals())
return gen_func # noqa
def add_list_endpoint(self, endpoint: MapdataEndpoint, tag: str):
view_params = self.common_params(endpoint)
Query # noqa
if endpoint.filters:
filters_name = endpoint.filters.__name__
view_params["filters"] = f"Query[{filters_name}]"
list_func = self._make_endpoint(
view_params=view_params,
call_func=mapdata_list_endpoint,
add_call_params={"model": endpoint.model.__name__}
)
list_func.__name__ = f"{endpoint.model_name}_list"
self.router.get(f"/{endpoint.model_name_plural}/", summary=f"{endpoint.model_name} list",
tags=[f"mapdata-{tag}"], description=schema_description(endpoint.schema),
response={200: list[endpoint.schema],
**(validate_responses if endpoint.filters else {}),
**auth_responses})(
api_etag()(list_func)
)
def add_by_id_endpoint(self, endpoint: MapdataEndpoint, tag: str):
view_params = self.common_params(endpoint)
PositiveInt # noqa
id_field = f"{endpoint.model_name}_id"
view_params[id_field] = "PositiveInt"
list_func = self._make_endpoint(
view_params=view_params,
call_func=mapdata_retrieve_endpoint,
add_call_params={"model": endpoint.model.__name__, "pk": id_field}
)
self.router.get(f'/{endpoint.model_name_plural}/{{{id_field}}}/', summary=f"{endpoint.model_name} by ID",
tags=[f"mapdata-{tag}"], description=schema_description(endpoint.schema),
response={200: endpoint.schema, **API404.dict(), **auth_responses})(
api_etag()(list_func)
)
class LevelFilters(ByGroupFilter, ByOnTopOfFilter):
pass
@mapdata_api_router.get('/levels/', summary="level list",
tags=["mapdata-root"], description=schema_description(LevelSchema),
response={200: list[LevelSchema], **validate_responses, **auth_responses})
@api_etag()
def level_list(request, filters: Query[LevelFilters]):
return mapdata_list_endpoint(request, model=Level, filters=filters)
@mapdata_api_router.get('/levels/{level_id}/', summary="level by ID",
tags=["mapdata-root"], description=schema_description(LevelSchema),
response={200: LevelSchema, **API404.dict(), **auth_responses})
@api_etag()
def level_by_id(request, level_id: int):
return mapdata_retrieve_endpoint(request, Level, pk=level_id)
"""
Buildings
"""
@mapdata_api_router.get('/buildings/', summary="building list",
tags=["mapdata-level"], description=schema_description(BuildingSchema),
response={200: list[BuildingSchema], **validate_responses, **auth_responses})
@api_etag(base_mapdata=True)
def building_list(request, filters: Query[LevelGeometryFilter]):
return mapdata_list_endpoint(request, model=Building, filters=filters)
@mapdata_api_router.get('/buildings/{building_id}/', summary="building by ID",
tags=["mapdata-level"], description=schema_description(BuildingSchema),
response={200: BuildingSchema, **API404.dict(), **auth_responses})
@api_etag(base_mapdata=True)
def building_by_id(request, building_id: int):
return mapdata_retrieve_endpoint(request, Building, pk=building_id)
"""
Spaces
"""
class SpaceFilters(ByGroupFilter, LevelGeometryFilter):
pass
@mapdata_api_router.get('/spaces/', summary="space list",
tags=["mapdata-level"], description=schema_description(SpaceSchema),
response={200: list[SpaceSchema], **validate_responses, **auth_responses})
@api_etag(base_mapdata=True)
def space_list(request, filters: Query[SpaceFilters]):
return mapdata_list_endpoint(request, model=Space, filters=filters)
@mapdata_api_router.get('/space/{space_id}/', summary="space by ID",
tags=["mapdata-level"], description=schema_description(SpaceSchema),
response={200: SpaceSchema, **API404.dict(), **auth_responses})
@api_etag(base_mapdata=True)
def space_by_id(request, space_id: int):
return mapdata_retrieve_endpoint(request, Space, pk=space_id)
"""
Doors
"""
@mapdata_api_router.get('/doors/', summary="door list",
tags=["mapdata-level"], description=schema_description(DoorSchema),
response={200: list[DoorSchema], **validate_responses, **auth_responses})
@api_etag(base_mapdata=True)
def door_list(request, filters: Query[LevelGeometryFilter]):
return mapdata_list_endpoint(request, model=Door, filters=filters)
@mapdata_api_router.get('/doors/{door_id}/', summary="door by ID",
tags=["mapdata-level"], description=schema_description(DoorSchema),
response={200: DoorSchema, **API404.dict(), **auth_responses})
@api_etag(base_mapdata=True)
def door_by_id(request, door_id: int):
return mapdata_retrieve_endpoint(request, Door, pk=door_id)
"""
Holes
"""
@mapdata_api_router.get('/holes/', summary="hole list",
tags=["mapdata-space"], description=schema_description(HoleSchema),
response={200: list[HoleSchema], **validate_responses, **auth_responses})
@api_etag()
def hole_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=Hole, filters=filters)
@mapdata_api_router.get('/holes/{hole_id}/', summary="hole by ID",
tags=["mapdata-space"], description=schema_description(HoleSchema),
response={200: HoleSchema, **API404.dict(), **auth_responses})
@api_etag()
def hole_by_id(request, hole_id: int):
return mapdata_retrieve_endpoint(request, Hole, pk=hole_id)
"""
Areas
"""
class AreaFilters(ByGroupFilter, SpaceGeometryFilter):
pass
@mapdata_api_router.get('/areas/', summary="area list",
tags=["mapdata-space"], description=schema_description(AreaSchema),
response={200: list[AreaSchema], **validate_responses, **auth_responses})
@api_etag()
def area_list(request, filters: Query[AreaFilters]):
return mapdata_list_endpoint(request, model=Area, filters=filters)
@mapdata_api_router.get('/areas/{area_id}/', summary="area by ID",
tags=["mapdata-space"], description=schema_description(AreaSchema),
response={200: AreaSchema, **API404.dict(), **auth_responses})
@api_etag()
def area_by_id(request, area_id: int):
return mapdata_retrieve_endpoint(request, Area, pk=area_id)
"""
Stairs
"""
@mapdata_api_router.get('/stairs/', summary="stair list",
tags=["mapdata-space"], description=schema_description(StairSchema),
response={200: list[StairSchema], **validate_responses, **auth_responses})
@api_etag()
def stair_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=Stair, filters=filters)
@mapdata_api_router.get('/stairs/{stair_id}/', summary="stair by ID",
tags=["mapdata-space"], description=schema_description(StairSchema),
response={200: StairSchema, **API404.dict(), **auth_responses})
@api_etag()
def stair_by_id(request, stair_id: int):
return mapdata_retrieve_endpoint(request, Stair, pk=stair_id)
"""
Ramps
"""
@mapdata_api_router.get('/ramps/', summary="ramp list",
tags=["mapdata-space"], description=schema_description(RampSchema),
response={200: list[RampSchema], **validate_responses, **auth_responses})
@api_etag()
def ramp_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=Ramp, filters=filters)
@mapdata_api_router.get('/ramps/{ramp_id}/', summary="ramp by ID",
tags=["mapdata-space"], description=schema_description(RampSchema),
response={200: RampSchema, **API404.dict(), **auth_responses})
@api_etag()
def ramp_by_id(request, ramp_id: int):
return mapdata_retrieve_endpoint(request, Ramp, pk=ramp_id)
"""
Obstacles
"""
@mapdata_api_router.get('/obstacles/', summary="obstacle list",
tags=["mapdata-space"], description=schema_description(ObstacleSchema),
response={200: list[ObstacleSchema], **validate_responses, **auth_responses})
@api_etag()
def obstacle_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=Obstacle, filters=filters)
@mapdata_api_router.get('/obstacles/{obstacle_id}/', summary="obstacle by ID",
tags=["mapdata-space"], description=schema_description(ObstacleSchema),
response={200: ObstacleSchema, **API404.dict(), **auth_responses})
@api_etag()
def obstacle_by_id(request, obstacle_id: int):
return mapdata_retrieve_endpoint(request, Obstacle, pk=obstacle_id)
"""
LineObstacles
"""
@mapdata_api_router.get('/lineobstacles/', summary="line obstacle list",
tags=["mapdata-space"], description=schema_description(LineObstacleSchema),
response={200: list[LineObstacleSchema], **validate_responses, **auth_responses})
@api_etag()
def lineobstacle_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=LineObstacle, filters=filters)
@mapdata_api_router.get('/lineobstacles/{lineobstacle_id}/', summary="line obstacle by ID",
tags=["mapdata-space"], description=schema_description(LineObstacleSchema),
response={200: LineObstacleSchema, **API404.dict(), **auth_responses},)
@api_etag()
def lineobstacle_by_id(request, lineobstacle_id: int):
return mapdata_retrieve_endpoint(request, LineObstacle, pk=lineobstacle_id)
"""
Columns
"""
@mapdata_api_router.get('/columns/', summary="column list",
tags=["mapdata-space"], description=schema_description(ColumnSchema),
response={200: list[ColumnSchema], **validate_responses, **auth_responses})
@api_etag()
def column_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=Column, filters=filters)
@mapdata_api_router.get('/columns/{column_id}/', summary="column by ID",
tags=["mapdata-space"], description=schema_description(ColumnSchema),
response={200: ColumnSchema, **API404.dict(), **auth_responses})
@api_etag()
def column_by_id(request, column_id: int):
return mapdata_retrieve_endpoint(request, Column, pk=column_id)
"""
POIs
"""
@mapdata_api_router.get('/pois/', summary="POI list",
tags=["mapdata-space"], description=schema_description(POISchema),
response={200: list[POISchema], **validate_responses, **auth_responses})
@api_etag()
def poi_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=POI, filters=filters)
@mapdata_api_router.get('/pois/{poi_id}/', summary="POI by ID",
tags=["mapdata-space"], description=schema_description(POISchema),
response={200: POISchema, **API404.dict(), **auth_responses})
@api_etag()
def poi_by_id(request, poi_id: int):
return mapdata_retrieve_endpoint(request, POI, pk=poi_id)
"""
LeaveDescriptions
"""
@mapdata_api_router.get('/leavedescriptions/', summary="leave description list",
tags=["mapdata-space"], description=schema_description(LeaveDescriptionSchema),
response={200: list[LeaveDescriptionSchema], **validate_responses, **auth_responses})
@api_etag()
def leavedescription_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=LeaveDescription, filters=filters)
@mapdata_api_router.get('/leavedescriptions/{leavedescription_id}/', summary="leave description by ID",
tags=["mapdata-space"], description=schema_description(LeaveDescriptionSchema),
response={200: LeaveDescriptionSchema, **API404.dict(), **auth_responses})
@api_etag()
def leavedescription_by_id(request, leavedescription_id: int):
return mapdata_retrieve_endpoint(request, LeaveDescription, pk=leavedescription_id)
"""
CrossDescriptions
"""
@mapdata_api_router.get('/crossdescriptions/', summary="cross description list",
tags=["mapdata-space"], description=schema_description(CrossDescriptionSchema),
response={200: list[CrossDescriptionSchema], **validate_responses, **auth_responses})
@api_etag()
def crossdescription_list(request, filters: Query[SpaceGeometryFilter]):
return mapdata_list_endpoint(request, model=CrossDescription, filters=filters)
@mapdata_api_router.get('/crossdescriptions/{crossdescription_id}/', summary="cross description by ID",
tags=["mapdata-space"], description=schema_description(CrossDescriptionSchema),
response={200: CrossDescriptionSchema, **API404.dict(), **auth_responses})
@api_etag()
def crossdescription_by_id(request, crossdescription_id: int):
return mapdata_retrieve_endpoint(request, CrossDescription, pk=crossdescription_id)
"""
LocationGroup
"""
@mapdata_api_router.get('/locationgroups/', summary="location group list",
tags=["mapdata-root"], description=schema_description(LocationGroupSchema),
response={200: list[LocationGroupSchema], **validate_responses, **auth_responses})
@api_etag()
def locationgroup_list(request, filters: Query[ByCategoryFilter]):
return mapdata_list_endpoint(request, model=LocationGroup, filters=filters)
@mapdata_api_router.get('/locationgroups/{locationgroup_id}/', summary="location group by ID",
tags=["mapdata-root"], description=schema_description(LocationGroupSchema),
response={200: LocationGroupSchema, **API404.dict(), **auth_responses})
@api_etag()
def locationgroup_by_id(request, locationgroup_id: int):
return mapdata_retrieve_endpoint(request, LocationGroup, pk=locationgroup_id)
"""
LocationGroupCategories
"""
@mapdata_api_router.get('/locationgroupcategories/', summary="location group category list",
tags=["mapdata-root"], description=schema_description(LocationGroupCategorySchema),
response={200: list[LocationGroupCategorySchema], **auth_responses})
@api_etag()
def locationgroupcategory_list(request):
return mapdata_list_endpoint(request, model=LocationGroupCategory)
@mapdata_api_router.get('/locationgroupcategories/{category_id}/', summary="location group category by ID",
tags=["mapdata-root"], description=schema_description(LocationGroupCategorySchema),
response={200: LocationGroupCategorySchema, **API404.dict(), **auth_responses})
@api_etag()
def locationgroupcategory_by_id(request, category_id: int):
return mapdata_retrieve_endpoint(request, LocationGroupCategory, pk=category_id)
"""
Sources
"""
@mapdata_api_router.get('/sources/', summary="source list",
tags=["mapdata-root"], description=schema_description(SourceSchema),
response={200: list[SourceSchema], **auth_responses})
@api_etag()
def source_list(request):
return mapdata_list_endpoint(request, model=Source)
@mapdata_api_router.get('/sources/{source_id}/', summary="source by ID",
tags=["mapdata-root"], description=schema_description(SourceSchema),
response={200: SourceSchema, **API404.dict(), **auth_responses})
@api_etag()
def source_by_id(request, source_id: int):
return mapdata_retrieve_endpoint(request, Source, pk=source_id)
"""
AccessRestrictions
"""
@mapdata_api_router.get('/accessrestrictions/', summary="access restriction list",
tags=["mapdata-root"], description=schema_description(AccessRestrictionSchema),
response={200: list[AccessRestrictionSchema], **auth_responses})
@api_etag()
def accessrestriction_list(request):
return mapdata_list_endpoint(request, model=AccessRestriction)
@mapdata_api_router.get('/accessrestrictions/{accessrestriction_id}/', summary="access restriction by ID",
tags=["mapdata-root"], description=schema_description(AccessRestrictionSchema),
response={200: AccessRestrictionSchema, **API404.dict(), **auth_responses})
@api_etag()
def accessrestriction_by_id(request, accessrestriction_id: int):
return mapdata_retrieve_endpoint(request, AccessRestriction, pk=accessrestriction_id)
"""
AccessRestrictionGroups
"""
@mapdata_api_router.get('/accessrestrictiongroups/', summary="access restriction group list",
tags=["mapdata-root"], description=schema_description(AccessRestrictionGroupSchema),
response={200: list[AccessRestrictionGroupSchema], **auth_responses})
@api_etag()
def accessrestrictiongroup_list(request):
return mapdata_list_endpoint(request, model=AccessRestrictionGroup)
@mapdata_api_router.get('/accessrestrictiongroups/{group_id}/', summary="access restriction group by ID",
tags=["mapdata-root"], description=schema_description(AccessRestrictionGroupSchema),
response={200: AccessRestrictionGroupSchema, **API404.dict(), **auth_responses})
@api_etag()
def accessrestrictiongroups_by_id(request, group_id: int):
return mapdata_retrieve_endpoint(request, AccessRestrictionGroup, pk=group_id)
"""
DynamicLocations
"""
@mapdata_api_router.get('/dynamiclocations/', summary="dynamic location list",
tags=["mapdata-root"], description=schema_description(DynamicLocationSchema),
response={200: list[DynamicLocationSchema], **auth_responses})
@api_etag()
def dynamiclocation_list(request):
return mapdata_list_endpoint(request, model=DynamicLocation)
@mapdata_api_router.get('/dynamiclocations/{dynamiclocation_id}/', summary="dynamic location by ID",
tags=["mapdata-root"], description=schema_description(DynamicLocationSchema),
response={200: DynamicLocationSchema, **API404.dict(), **auth_responses})
@api_etag()
def dynamiclocation_by_id(request, dynamiclocation_id: int):
return mapdata_retrieve_endpoint(request, DynamicLocation, pk=dynamiclocation_id)
mapdata_endpoints: dict[str, list[MapdataEndpoint]] = {
"root": [
MapdataEndpoint(
model=Level,
schema=LevelSchema,
filters=LevelFilters
),
MapdataEndpoint(
model=LocationGroup,
schema=LocationGroupSchema,
filters=ByCategoryFilter,
),
MapdataEndpoint(
model=LocationGroupCategory,
schema=LocationGroupCategorySchema,
),
MapdataEndpoint(
model=Source,
schema=SourceSchema,
),
MapdataEndpoint(
model=AccessRestriction,
schema=AccessRestrictionSchema,
),
MapdataEndpoint(
model=AccessRestrictionGroup,
schema=AccessRestrictionGroupSchema,
),
MapdataEndpoint(
model=DynamicLocation,
schema=DynamicLocationSchema,
),
],
"level": [
MapdataEndpoint(
model=Building,
schema=BuildingSchema,
filters=LevelGeometryFilter
),
MapdataEndpoint(
model=Space,
schema=SpaceSchema,
filters=SpaceFilters,
),
MapdataEndpoint(
model=Door,
schema=DoorSchema,
filters=LevelGeometryFilter,
)
],
"space": [
MapdataEndpoint(
model=Hole,
schema=HoleSchema,
filters=SpaceGeometryFilter,
),
MapdataEndpoint(
model=Area,
schema=AreaSchema,
filters=AreaFilters,
),
MapdataEndpoint(
model=Stair,
schema=StairSchema,
filters=SpaceGeometryFilter,
),
MapdataEndpoint(
model=Ramp,
schema=RampSchema,
filters=SpaceGeometryFilter,
),
MapdataEndpoint(
model=Obstacle,
schema=ObstacleSchema,
filters=SpaceGeometryFilter,
),
MapdataEndpoint(
model=LineObstacle,
schema=LineObstacleSchema,
filters=SpaceGeometryFilter,
),
MapdataEndpoint(
model=Column,
schema=ColumnSchema,
filters=SpaceGeometryFilter,
),
MapdataEndpoint(
model=POI,
schema=POISchema,
filters=SpaceGeometryFilter,
),
MapdataEndpoint(
model=LeaveDescription,
schema=LeaveDescriptionSchema,
filters=BySpaceFilter,
),
MapdataEndpoint(
model=CrossDescription,
schema=CrossDescriptionSchema,
filters=BySpaceFilter,
),
],
}
MapdataAPIBuilder(router=mapdata_api_router).build_all_endpoints(mapdata_endpoints)
"""
@ -495,6 +277,7 @@ Data overlays
"""
# todo: this wants to move into a MapDataEndpoint
@mapdata_api_router.get('/overlays/', summary="data overlay list",
tags=["mapdata-root"], description=schema_description(DynamicLocationSchema),
response={200: list[DataOverlaySchema], **auth_responses})
@ -503,6 +286,7 @@ def dataoverlay_list(request):
return mapdata_list_endpoint(request, model=DataOverlay)
# todo: this wants to move into a MapDataEndpoint
@mapdata_api_router.get('/overlays/{overlay_id}/', summary="features for overlay by overlay ID",
tags=["mapdata-root"], description=schema_description(DynamicLocationSchema),
response={200: list[DataOverlayFeatureSchema], **API404.dict(), **auth_responses})