From f70040fe1d04a2e2a5eb7e62e463f5da61a811f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Laura=20Kl=C3=BCnder?= Date: Mon, 2 Dec 2024 23:40:41 +0100 Subject: [PATCH] auto-generate mapdata api endpoints --- src/c3nav/mapdata/api/mapdata.py | 620 ++++++++++--------------------- 1 file changed, 202 insertions(+), 418 deletions(-) diff --git a/src/c3nav/mapdata/api/mapdata.py b/src/c3nav/mapdata/api/mapdata.py index b24d2e76..86ea22f4 100644 --- a/src/c3nav/mapdata/api/mapdata.py +++ b/src/c3nav/mapdata/api/mapdata.py @@ -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})