add tons of mapdata endpoints, enhance documentation and operation IDs
This commit is contained in:
parent
7447537c4a
commit
f43d458fc4
12 changed files with 637 additions and 169 deletions
|
@ -65,6 +65,6 @@ class APITokenAuth(HttpBearer):
|
|||
return user
|
||||
|
||||
|
||||
validate_responses = {422: APIErrorSchema,}
|
||||
auth_responses = {401: APIErrorSchema,}
|
||||
auth_permission_responses = {401: APIErrorSchema, 403: APIErrorSchema,}
|
||||
validate_responses = {422: APIErrorSchema, }
|
||||
auth_responses = {401: APIErrorSchema, }
|
||||
auth_permission_responses = {401: APIErrorSchema, 403: APIErrorSchema, }
|
||||
|
|
69
src/c3nav/api/ninja.py
Normal file
69
src/c3nav/api/ninja.py
Normal file
|
@ -0,0 +1,69 @@
|
|||
from ninja import NinjaAPI, Swagger
|
||||
from ninja.operation import Operation
|
||||
from ninja.schema import NinjaGenerateJsonSchema
|
||||
|
||||
from c3nav.api.exceptions import CustomAPIException
|
||||
from c3nav.api.newauth import APITokenAuth
|
||||
|
||||
|
||||
class c3navAPI(NinjaAPI):
|
||||
def get_openapi_operation_id(self, operation: Operation) -> str:
|
||||
name = operation.view_func.__name__
|
||||
result = f"c3nav_{operation.tags[0]}_{name}"
|
||||
return result
|
||||
|
||||
|
||||
description = """
|
||||
Nearly all endpoints require authentication, but guest authentication can be used.
|
||||
|
||||
API endpoints may change to add more features and properties,
|
||||
but no properties will be removed without a version change.
|
||||
""".strip()
|
||||
ninja_api = c3navAPI(
|
||||
title="c3nav API",
|
||||
version="v2",
|
||||
description=description,
|
||||
|
||||
docs_url="/",
|
||||
docs=Swagger(settings={
|
||||
"persistAuthorization": True,
|
||||
"defaultModelRendering": "model",
|
||||
}),
|
||||
|
||||
auth=APITokenAuth(),
|
||||
|
||||
openapi_extra={
|
||||
"tags": [
|
||||
{
|
||||
"name": "auth",
|
||||
"description": "Get and manage API access",
|
||||
},
|
||||
{
|
||||
"name": "map",
|
||||
"description": "Common map endpoints",
|
||||
},
|
||||
{
|
||||
"name": "mapdata",
|
||||
"description": "Access the raw map data",
|
||||
},
|
||||
{
|
||||
"name": "mesh",
|
||||
"description": "Manage the location node mesh network",
|
||||
},
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
"""
|
||||
ugly hack: remove schema from the end of definition names
|
||||
"""
|
||||
orig_normalize_name = NinjaGenerateJsonSchema.normalize_name
|
||||
def wrap_normalize_name(self, name: str): # noqa
|
||||
return orig_normalize_name(self, name).removesuffix('Schema')
|
||||
NinjaGenerateJsonSchema.normalize_name = wrap_normalize_name # noqa
|
||||
|
||||
|
||||
@ninja_api.exception_handler(CustomAPIException)
|
||||
def on_invalid_token(request, exc):
|
||||
return ninja_api.create_response(request, {"detail": exc.detail}, status=exc.status_code)
|
|
@ -16,7 +16,30 @@ class APIErrorSchema(Schema):
|
|||
|
||||
|
||||
class PolygonSchema(Schema):
|
||||
"""
|
||||
A GeoJSON Polygon
|
||||
"""
|
||||
type: Literal["Polygon"]
|
||||
coordinates: list[list[tuple[float, float]]] = APIField(
|
||||
example=[[1.5, 1.5], [1.5, 2.5], [2.5, 2.5], [2.5, 2.5]]
|
||||
example=[[[1.5, 1.5], [1.5, 2.5], [2.5, 2.5], [2.5, 2.5]]]
|
||||
)
|
||||
|
||||
|
||||
class LineStringSchema(Schema):
|
||||
"""
|
||||
A GeoJSON LineString
|
||||
"""
|
||||
type: Literal["LineString"]
|
||||
coordinates: list[tuple[float, float]] = APIField(
|
||||
example=[[1.5, 1.5], [2.5, 2.5], [5, 8.7]]
|
||||
)
|
||||
|
||||
|
||||
class PointSchema(Schema):
|
||||
"""
|
||||
A GeoJSON Point
|
||||
"""
|
||||
type: Literal["Point"]
|
||||
coordinates: tuple[float, float] = APIField(
|
||||
example=[1, 2.5]
|
||||
)
|
||||
|
|
|
@ -2,19 +2,15 @@ import inspect
|
|||
import re
|
||||
from collections import OrderedDict
|
||||
|
||||
from django.conf import settings
|
||||
from django.urls import include, path, re_path
|
||||
from django.utils.functional import cached_property
|
||||
from ninja import NinjaAPI, Swagger
|
||||
from ninja.schema import NinjaGenerateJsonSchema
|
||||
from rest_framework.generics import GenericAPIView
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.routers import SimpleRouter
|
||||
|
||||
from c3nav.api.api import SessionViewSet
|
||||
from c3nav.api.exceptions import CustomAPIException
|
||||
from c3nav.api.newapi import auth_api_router
|
||||
from c3nav.api.newauth import APITokenAuth
|
||||
from c3nav.api.ninja import ninja_api
|
||||
from c3nav.editor.api import ChangeSetViewSet, EditorViewSet
|
||||
from c3nav.mapdata.api import (AccessRestrictionGroupViewSet, AccessRestrictionViewSet, AreaViewSet, BuildingViewSet,
|
||||
ColumnViewSet, CrossDescriptionViewSet, DoorViewSet, DynamicLocationPositionViewSet,
|
||||
|
@ -22,70 +18,27 @@ from c3nav.mapdata.api import (AccessRestrictionGroupViewSet, AccessRestrictionV
|
|||
LocationBySlugViewSet, LocationGroupCategoryViewSet, LocationGroupViewSet,
|
||||
LocationViewSet, MapViewSet, ObstacleViewSet, POIViewSet, RampViewSet, SourceViewSet,
|
||||
SpaceViewSet, StairViewSet, UpdatesViewSet)
|
||||
from c3nav.mapdata.newapi.endpoints import map_api_router
|
||||
from c3nav.mapdata.newapi.map import map_api_router
|
||||
from c3nav.mapdata.newapi.mapdata import mapdata_api_router
|
||||
from c3nav.mapdata.utils.user import can_access_editor
|
||||
from c3nav.mesh.api import FirmwareViewSet
|
||||
from c3nav.mesh.newapi import mesh_api_router
|
||||
from c3nav.routing.api import RoutingViewSet
|
||||
|
||||
description = """
|
||||
Nearly all endpoints require authentication, but guest authentication can be used.
|
||||
|
||||
API endpoints may change to add more features and properties,
|
||||
but no properties will be removed without a version change.
|
||||
""".strip()
|
||||
|
||||
ninja_api = NinjaAPI(
|
||||
title="c3nav API",
|
||||
version="v2",
|
||||
description=description,
|
||||
|
||||
docs_url="/",
|
||||
docs=Swagger(settings={
|
||||
"persistAuthorization": True,
|
||||
"defaultModelRendering": "model",
|
||||
}),
|
||||
|
||||
auth=APITokenAuth(),
|
||||
|
||||
openapi_extra={
|
||||
"tags": [
|
||||
{
|
||||
"name": "auth",
|
||||
"description": "Get and manage API access",
|
||||
},
|
||||
{
|
||||
"name": "map",
|
||||
"description": "Access the map data",
|
||||
},
|
||||
{
|
||||
"name": "mesh",
|
||||
"description": "Manage the location node mesh network",
|
||||
},
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@ninja_api.exception_handler(CustomAPIException)
|
||||
def on_invalid_token(request, exc):
|
||||
return ninja_api.create_response(request, {"detail": exc.detail}, status=exc.status_code)
|
||||
|
||||
|
||||
"""
|
||||
ugly hack: remove schema from the end of definition names
|
||||
new API (v2)
|
||||
"""
|
||||
orig_normalize_name = NinjaGenerateJsonSchema.normalize_name
|
||||
def wrap_normalize_name(self, name: str): # noqa
|
||||
return orig_normalize_name(self, name).removesuffix('Schema')
|
||||
NinjaGenerateJsonSchema.normalize_name = wrap_normalize_name # noqa
|
||||
|
||||
|
||||
ninja_api.add_router("/auth/", auth_api_router)
|
||||
ninja_api.add_router("/map/", map_api_router)
|
||||
ninja_api.add_router("/mapdata/", mapdata_api_router)
|
||||
ninja_api.add_router("/mesh/", mesh_api_router)
|
||||
|
||||
|
||||
"""
|
||||
legacy API
|
||||
"""
|
||||
router = SimpleRouter()
|
||||
|
||||
router.register(r'map', MapViewSet, basename='map')
|
||||
router.register(r'levels', LevelViewSet)
|
||||
router.register(r'buildings', BuildingViewSet)
|
||||
|
|
|
@ -44,6 +44,12 @@ class SpaceGeometryMixin(GeometryMixin):
|
|||
result['opacity'] = self.opacity
|
||||
return result
|
||||
|
||||
def _serialize(self, space=True, **kwargs):
|
||||
result = super()._serialize(**kwargs)
|
||||
if space:
|
||||
result['space'] = self.space_id
|
||||
return result
|
||||
|
||||
@property
|
||||
def subtitle(self):
|
||||
base_subtitle = super().subtitle
|
||||
|
@ -126,6 +132,7 @@ class Area(SpaceGeometryMixin, SpecificLocation, models.Model):
|
|||
|
||||
def _serialize(self, **kwargs):
|
||||
result = super()._serialize(**kwargs)
|
||||
result["slow_down_factor"] = float(self.slow_down_factor)
|
||||
return result
|
||||
|
||||
@property
|
||||
|
|
|
@ -1,92 +0,0 @@
|
|||
from typing import Optional, Sequence, Type
|
||||
|
||||
from django.db.models import Model
|
||||
from ninja import Query
|
||||
from ninja import Router as APIRouter
|
||||
from ninja.pagination import paginate
|
||||
|
||||
from c3nav.api.exceptions import API404
|
||||
from c3nav.api.newauth import auth_responses
|
||||
from c3nav.mapdata.api import optimize_query
|
||||
from c3nav.mapdata.models import Level, Source
|
||||
from c3nav.mapdata.models.access import AccessPermission
|
||||
from c3nav.mapdata.schemas.filters import FilterSchema, GroupFilter, OnTopOfFilter
|
||||
from c3nav.mapdata.schemas.models import LevelSchema
|
||||
from c3nav.mapdata.schemas.responses import BoundsSchema
|
||||
|
||||
map_api_router = APIRouter(tags=["map"])
|
||||
|
||||
|
||||
@map_api_router.get('/bounds/', summary="Get map boundaries",
|
||||
response={200: BoundsSchema, **auth_responses})
|
||||
def bounds(request):
|
||||
return {
|
||||
"bounds": Source.max_bounds(),
|
||||
}
|
||||
|
||||
|
||||
def mapdata_list_endpoint(request,
|
||||
model: Type[Model],
|
||||
filters: Optional[FilterSchema] = None,
|
||||
order_by: Sequence[str] = ('pk',)):
|
||||
# todo: request permissions based on api key
|
||||
|
||||
# todo: pagination cache?
|
||||
|
||||
# generate cache_key
|
||||
# todo: don't ignore request language
|
||||
cache_key = 'mapdata:api:%s:%s' % (model.__name__, AccessPermission.cache_key_for_request(request))
|
||||
if filters:
|
||||
for name in filters.model_fields_set: # noqa
|
||||
value = getattr(filters, name)
|
||||
if value is None:
|
||||
continue
|
||||
cache_key += ';%s,%s' % (name, value)
|
||||
|
||||
# todo: we have the cache key, this would be a great time for a shortcut
|
||||
|
||||
# validate filters
|
||||
if filters:
|
||||
filters.validate(request)
|
||||
|
||||
# get the queryset and filter it
|
||||
qs = optimize_query(
|
||||
model.qs_for_request(request) if hasattr(model, 'qs_for_request') else model.objects.all()
|
||||
)
|
||||
if filters:
|
||||
qs = filters.filter_qs(qs)
|
||||
|
||||
# order_by
|
||||
qs = qs.order_by(*order_by)
|
||||
|
||||
# todo: can access geometry… using defer?
|
||||
|
||||
return qs
|
||||
|
||||
|
||||
def mapdata_retrieve_endpoint(request, model: Type[Model], **lookups):
|
||||
try:
|
||||
return optimize_query(
|
||||
model.qs_for_request(request) if hasattr(model, 'qs_for_request') else model.objects.all()
|
||||
).get(**lookups)
|
||||
except model.DoesNotExist:
|
||||
raise API404("%s not found" % model.__name__.lower())
|
||||
|
||||
|
||||
class LevelFilters(GroupFilter, OnTopOfFilter):
|
||||
pass
|
||||
|
||||
|
||||
@map_api_router.get('/levels/', response=list[LevelSchema],
|
||||
summary="Get level list")
|
||||
@paginate
|
||||
def levels_list(request, filters: Query[LevelFilters]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Level, filters=filters)
|
||||
|
||||
|
||||
@map_api_router.get('/levels/{level_id}/', response=LevelSchema,
|
||||
summary="Get level by ID")
|
||||
def level_detail(request, level_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Level, pk=level_id)
|
15
src/c3nav/mapdata/newapi/map.py
Normal file
15
src/c3nav/mapdata/newapi/map.py
Normal file
|
@ -0,0 +1,15 @@
|
|||
from ninja import Router as APIRouter
|
||||
|
||||
from c3nav.api.newauth import auth_responses
|
||||
from c3nav.mapdata.models import Source
|
||||
from c3nav.mapdata.schemas.responses import BoundsSchema
|
||||
|
||||
map_api_router = APIRouter(tags=["map"])
|
||||
|
||||
|
||||
@map_api_router.get('/bounds/', summary="Get map boundaries",
|
||||
response={200: BoundsSchema, **auth_responses})
|
||||
def bounds(request):
|
||||
return {
|
||||
"bounds": Source.max_bounds(),
|
||||
}
|
318
src/c3nav/mapdata/newapi/mapdata.py
Normal file
318
src/c3nav/mapdata/newapi/mapdata.py
Normal file
|
@ -0,0 +1,318 @@
|
|||
from typing import Optional, Sequence, Type
|
||||
|
||||
from django.db.models import Model
|
||||
from ninja import Query
|
||||
from ninja import Router as APIRouter
|
||||
from ninja.pagination import paginate
|
||||
|
||||
from c3nav.api.exceptions import API404
|
||||
from c3nav.mapdata.api import optimize_query
|
||||
from c3nav.mapdata.models import Area, Building, Door, Hole, Level, Space, Stair
|
||||
from c3nav.mapdata.models.access import AccessPermission
|
||||
from c3nav.mapdata.models.geometry.space import POI, Column, LineObstacle, Obstacle, Ramp
|
||||
from c3nav.mapdata.schemas.filters import ByGroupFilter, ByLevelFilter, ByOnTopOfFilter, BySpaceFilter, FilterSchema
|
||||
from c3nav.mapdata.schemas.models import (AreaSchema, BuildingSchema, ColumnSchema, DoorSchema, HoleSchema, LevelSchema,
|
||||
LineObstacleSchema, ObstacleSchema, POISchema, RampSchema, SpaceSchema,
|
||||
StairSchema)
|
||||
|
||||
mapdata_api_router = APIRouter(tags=["mapdata"])
|
||||
|
||||
|
||||
def mapdata_list_endpoint(request,
|
||||
model: Type[Model],
|
||||
filters: Optional[FilterSchema] = None,
|
||||
order_by: Sequence[str] = ('pk',)):
|
||||
# todo: request permissions based on api key
|
||||
|
||||
# todo: pagination cache?
|
||||
|
||||
# generate cache_key
|
||||
# todo: don't ignore request language
|
||||
cache_key = 'mapdata:api:%s:%s' % (model.__name__, AccessPermission.cache_key_for_request(request))
|
||||
if filters:
|
||||
for name in filters.model_fields_set: # noqa
|
||||
value = getattr(filters, name)
|
||||
if value is None:
|
||||
continue
|
||||
cache_key += ';%s,%s' % (name, value)
|
||||
|
||||
# todo: we have the cache key, this would be a great time for a shortcut
|
||||
|
||||
# validate filters
|
||||
if filters:
|
||||
filters.validate(request)
|
||||
|
||||
# get the queryset and filter it
|
||||
qs = optimize_query(
|
||||
model.qs_for_request(request) if hasattr(model, 'qs_for_request') else model.objects.all()
|
||||
)
|
||||
if filters:
|
||||
qs = filters.filter_qs(qs)
|
||||
|
||||
# order_by
|
||||
qs = qs.order_by(*order_by)
|
||||
|
||||
# todo: can access geometry… using defer?
|
||||
|
||||
return qs
|
||||
|
||||
|
||||
def mapdata_retrieve_endpoint(request, model: Type[Model], **lookups):
|
||||
try:
|
||||
return optimize_query(
|
||||
model.qs_for_request(request) if hasattr(model, 'qs_for_request') else model.objects.all()
|
||||
).get(**lookups)
|
||||
except model.DoesNotExist:
|
||||
raise API404("%s not found" % model.__name__.lower())
|
||||
|
||||
|
||||
"""
|
||||
Levels
|
||||
"""
|
||||
|
||||
|
||||
class LevelFilters(ByGroupFilter, ByOnTopOfFilter):
|
||||
pass
|
||||
|
||||
|
||||
@mapdata_api_router.get('/levels/', response=list[LevelSchema],
|
||||
summary="Get level list")
|
||||
@paginate
|
||||
def level_list(request, filters: Query[LevelFilters]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Level, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/levels/{level_id}/', response=LevelSchema,
|
||||
summary="Get level by ID")
|
||||
def level_detail(request, level_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Level, pk=level_id)
|
||||
|
||||
|
||||
"""
|
||||
Buildings
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/buildings/', response=list[BuildingSchema],
|
||||
summary="Get building list")
|
||||
@paginate
|
||||
def building_list(request, filters: Query[ByLevelFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Building, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/buildings/{building_id}/', response=BuildingSchema,
|
||||
summary="Get building by ID")
|
||||
def building_detail(request, building_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Building, pk=building_id)
|
||||
|
||||
|
||||
"""
|
||||
Spaces
|
||||
"""
|
||||
|
||||
|
||||
class SpaceFilters(ByGroupFilter, ByLevelFilter):
|
||||
pass
|
||||
|
||||
|
||||
@mapdata_api_router.get('/spaces/', response=list[SpaceSchema],
|
||||
summary="Get space list")
|
||||
@paginate
|
||||
def space_list(request, filters: Query[SpaceFilters]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Space, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/space/{space_id}/', response=SpaceSchema,
|
||||
summary="Get space by ID")
|
||||
def space_detail(request, space_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Space, pk=space_id)
|
||||
|
||||
|
||||
"""
|
||||
Doors
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/doors/', response=list[DoorSchema],
|
||||
summary="Get door list")
|
||||
@paginate
|
||||
def door_list(request, filters: Query[ByLevelFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Door, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/doors/{door_id}/', response=DoorSchema,
|
||||
summary="Get door by ID")
|
||||
def door_detail(request, door_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Door, pk=door_id)
|
||||
|
||||
|
||||
"""
|
||||
Holes
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/holes/', response=list[HoleSchema],
|
||||
summary="Get hole list")
|
||||
@paginate
|
||||
def hole_list(request, filters: Query[BySpaceFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Hole, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/holes/{hole_id}/', response=HoleSchema,
|
||||
summary="Get hole by ID")
|
||||
def hole_detail(request, hole_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Hole, pk=hole_id)
|
||||
|
||||
|
||||
"""
|
||||
Areas
|
||||
"""
|
||||
|
||||
|
||||
class AreaFilters(ByGroupFilter, BySpaceFilter):
|
||||
pass
|
||||
|
||||
|
||||
@mapdata_api_router.get('/areas/', response=list[AreaSchema],
|
||||
summary="Get area list")
|
||||
@paginate
|
||||
def area_list(request, filters: Query[AreaFilters]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Area, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/areas/{area_id}/', response=AreaSchema,
|
||||
summary="Get area by ID")
|
||||
def area_detail(request, area_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Area, pk=area_id)
|
||||
|
||||
|
||||
"""
|
||||
Stairs
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/stairs/', response=list[StairSchema],
|
||||
summary="Get stair list")
|
||||
@paginate
|
||||
def stair_list(request, filters: Query[BySpaceFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Stair, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/stairs/{stair_id}/', response=StairSchema,
|
||||
summary="Get stair by ID")
|
||||
def stair_detail(request, stair_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Stair, pk=stair_id)
|
||||
|
||||
|
||||
"""
|
||||
Ramps
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/ramps/', response=list[RampSchema],
|
||||
summary="Get ramp list")
|
||||
@paginate
|
||||
def ramp_list(request, filters: Query[BySpaceFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Ramp, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/ramps/{ramp_id}/', response=RampSchema,
|
||||
summary="Get ramp by ID")
|
||||
def ramp_detail(request, ramp_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Ramp, pk=ramp_id)
|
||||
|
||||
|
||||
"""
|
||||
Obstacles
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/obstacles/', response=list[ObstacleSchema],
|
||||
summary="Get obstacle list")
|
||||
@paginate
|
||||
def obstacle_list(request, filters: Query[BySpaceFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Obstacle, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/obstacles/{obstacle_id}/', response=ObstacleSchema,
|
||||
summary="Get obstacle by ID")
|
||||
def obstacle_detail(request, obstacle_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Obstacle, pk=obstacle_id)
|
||||
|
||||
|
||||
"""
|
||||
LineObstacles
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/lineobstacles/', response=list[LineObstacleSchema],
|
||||
summary="Get line obstacle list")
|
||||
@paginate
|
||||
def lineobstacle_list(request, filters: Query[BySpaceFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=LineObstacle, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/lineobstacles/{lineobstacle_id}/', response=LineObstacleSchema,
|
||||
summary="Get line obstacle by ID")
|
||||
def lineobstacle_detail(request, lineobstacle_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, LineObstacle, pk=lineobstacle_id)
|
||||
|
||||
|
||||
"""
|
||||
Columns
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/columns/', response=list[ColumnSchema],
|
||||
summary="Get column list")
|
||||
@paginate
|
||||
def column_list(request, filters: Query[BySpaceFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=Column, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/columns/{column_id}/', response=ColumnSchema,
|
||||
summary="Get column by ID")
|
||||
def column_detail(request, column_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, Column, pk=column_id)
|
||||
|
||||
|
||||
"""
|
||||
POIs
|
||||
"""
|
||||
|
||||
|
||||
@mapdata_api_router.get('/pois/', response=list[POISchema],
|
||||
summary="Get POI list")
|
||||
@paginate
|
||||
def poi_list(request, filters: Query[BySpaceFilter]):
|
||||
# todo cache?
|
||||
return mapdata_list_endpoint(request, model=POI, filters=filters)
|
||||
|
||||
|
||||
@mapdata_api_router.get('/pois/{poi_id}/', response=POISchema,
|
||||
summary="Get POI by ID")
|
||||
def poi_detail(request, poi_id: int):
|
||||
# todo: access, caching, filtering, etc
|
||||
return mapdata_retrieve_endpoint(request, POI, pk=poi_id)
|
|
@ -6,7 +6,7 @@ from ninja import Schema
|
|||
from pydantic import Field as APIField
|
||||
|
||||
from c3nav.api.exceptions import APIRequestValidationFailed
|
||||
from c3nav.mapdata.models import Level, LocationGroup, MapUpdate
|
||||
from c3nav.mapdata.models import Level, LocationGroup, MapUpdate, Space
|
||||
from c3nav.mapdata.models.access import AccessPermission
|
||||
|
||||
|
||||
|
@ -46,11 +46,47 @@ class FilterSchema(Schema):
|
|||
pass
|
||||
|
||||
|
||||
class GroupFilter(FilterSchema):
|
||||
class ByLevelFilter(FilterSchema):
|
||||
level: Optional[int] = APIField(
|
||||
None,
|
||||
title="filter by level",
|
||||
description="if set, only items belonging to the level with this ID will be shown"
|
||||
)
|
||||
|
||||
def validate(self, request):
|
||||
super().validate(request)
|
||||
if self.level is not None:
|
||||
assert_valid_value(request, Level, "pk", {self.level})
|
||||
|
||||
def filter_qs(self, qs: QuerySet) -> QuerySet:
|
||||
if self.level is not None:
|
||||
qs = qs.filter(level_id=self.level)
|
||||
return super().filter_qs(qs)
|
||||
|
||||
|
||||
class BySpaceFilter(FilterSchema):
|
||||
space: Optional[int] = APIField(
|
||||
None,
|
||||
title="filter by space",
|
||||
description="if set, only items belonging to the space with this ID will be shown"
|
||||
)
|
||||
|
||||
def validate(self, request):
|
||||
super().validate(request)
|
||||
if self.space is not None:
|
||||
assert_valid_value(request, Space, "pk", {self.space})
|
||||
|
||||
def filter_qs(self, qs: QuerySet) -> QuerySet:
|
||||
if self.space is not None:
|
||||
qs = qs.filter(groups=self.space)
|
||||
return super().filter_qs(qs)
|
||||
|
||||
|
||||
class ByGroupFilter(FilterSchema):
|
||||
group: Optional[int] = APIField(
|
||||
None,
|
||||
title="filter by location group",
|
||||
description="if set, only items belonging to the location group with that ID will be shown"
|
||||
description="if set, only items belonging to the location group with this ID will be shown"
|
||||
)
|
||||
|
||||
def validate(self, request):
|
||||
|
@ -59,13 +95,12 @@ class GroupFilter(FilterSchema):
|
|||
assert_valid_value(request, LocationGroup, "pk", {self.group})
|
||||
|
||||
def filter_qs(self, qs: QuerySet) -> QuerySet:
|
||||
qs = super().filter_qs(qs)
|
||||
if self.group is not None:
|
||||
qs = qs.filter(groups=self.group)
|
||||
return super().filter_qs(qs)
|
||||
|
||||
|
||||
class OnTopOfFilter(FilterSchema):
|
||||
class ByOnTopOfFilter(FilterSchema):
|
||||
on_top_of: Optional[Literal["null"] | int] = APIField(
|
||||
None,
|
||||
title='filter by on top of level ID (or "null")',
|
||||
|
|
|
@ -6,6 +6,7 @@ from pydantic import PositiveInt, model_validator
|
|||
from pydantic.functional_validators import ModelWrapValidatorHandler
|
||||
from pydantic_core.core_schema import ValidationInfo
|
||||
|
||||
from c3nav.api.schema import LineStringSchema, PointSchema, PolygonSchema
|
||||
from c3nav.api.utils import NonEmptyStr
|
||||
|
||||
|
||||
|
@ -93,7 +94,7 @@ class LabelSettingsSchema(TitledSchema, DjangoModelSchema):
|
|||
)
|
||||
|
||||
|
||||
class SpecificLocationSchema(LocationSchema, DjangoModelSchema):
|
||||
class SpecificLocationSchema(LocationSchema):
|
||||
grid_square: Optional[NonEmptyStr] = APIField(
|
||||
default=None,
|
||||
title="grid square",
|
||||
|
@ -121,3 +122,35 @@ class SpecificLocationSchema(LocationSchema, DjangoModelSchema):
|
|||
title="label override (preferred language)",
|
||||
description="preferred language based on the Accept-Language header."
|
||||
)
|
||||
|
||||
|
||||
class WithPolygonGeometrySchema(Schema):
|
||||
geometry: PolygonSchema = APIField(
|
||||
title="geometry",
|
||||
)
|
||||
|
||||
|
||||
class WithLineStringGeometrySchema(Schema):
|
||||
geometry: LineStringSchema = APIField(
|
||||
title="geometry",
|
||||
)
|
||||
|
||||
|
||||
class WithPointGeometrySchema(Schema):
|
||||
geometry: PointSchema = APIField(
|
||||
title="geometry",
|
||||
)
|
||||
|
||||
|
||||
class WithLevelSchema(SerializableSchema):
|
||||
level: PositiveInt = APIField(
|
||||
title="level",
|
||||
description="level id this object belongs to.",
|
||||
)
|
||||
|
||||
|
||||
class WithSpaceSchema(SerializableSchema):
|
||||
space: PositiveInt = APIField(
|
||||
title="space",
|
||||
description="space id this object belongs to.",
|
||||
)
|
||||
|
|
|
@ -1,14 +1,16 @@
|
|||
from typing import Optional
|
||||
|
||||
from ninja import Schema, Swagger
|
||||
from pydantic import Field as APIField
|
||||
from pydantic import PositiveFloat, PositiveInt
|
||||
from pydantic import NonNegativeFloat, PositiveFloat, PositiveInt
|
||||
from pydantic.color import Color
|
||||
|
||||
from c3nav.api.utils import NonEmptyStr
|
||||
from c3nav.mapdata.schemas.model_base import SpecificLocationSchema
|
||||
from c3nav.mapdata.schemas.model_base import (AccessRestrictionSchema, DjangoModelSchema, SpecificLocationSchema,
|
||||
WithLevelSchema, WithLineStringGeometrySchema, WithPointGeometrySchema,
|
||||
WithPolygonGeometrySchema, WithSpaceSchema)
|
||||
|
||||
|
||||
class LevelSchema(SpecificLocationSchema):
|
||||
class LevelSchema(SpecificLocationSchema, DjangoModelSchema):
|
||||
"""
|
||||
A physical level of the map, containing building, spaces, doors…
|
||||
|
||||
|
@ -32,5 +34,113 @@ class LevelSchema(SpecificLocationSchema):
|
|||
title="door height",
|
||||
)
|
||||
|
||||
class Config(Schema.Config):
|
||||
title = "Level"
|
||||
|
||||
class BuildingSchema(WithPolygonGeometrySchema, WithLevelSchema, DjangoModelSchema):
|
||||
"""
|
||||
A non-outdoor part of the map.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class SpaceSchema(WithPolygonGeometrySchema, SpecificLocationSchema, WithLevelSchema, DjangoModelSchema):
|
||||
"""
|
||||
An accessible area on a level. It can be outside-only or inside-only.
|
||||
|
||||
A space is a specific location, and can therefor be routed to and from, as well as belong to location groups.
|
||||
"""
|
||||
outside: bool = APIField(
|
||||
title="outside only",
|
||||
description="determines whether to truncate to buildings or to the outside of buildings"
|
||||
)
|
||||
height: Optional[PositiveFloat] = APIField(
|
||||
title="ceiling height",
|
||||
description="if not set, default height for this level will be used"
|
||||
)
|
||||
|
||||
|
||||
class DoorSchema(WithPolygonGeometrySchema, AccessRestrictionSchema, WithLevelSchema, DjangoModelSchema):
|
||||
"""
|
||||
A link between two spaces
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class HoleSchema(WithPolygonGeometrySchema, WithSpaceSchema):
|
||||
"""
|
||||
A hole in a space, showing the levels below
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class AreaSchema(WithPolygonGeometrySchema, SpecificLocationSchema, WithSpaceSchema, DjangoModelSchema):
|
||||
"""
|
||||
An area inside a space.
|
||||
|
||||
An area is a specific location, and can therefor be routed to and from, as well as belong to location groups.
|
||||
"""
|
||||
slow_down_factor: PositiveFloat = APIField(
|
||||
title="slow-down factor",
|
||||
description="how much walking in this area is slowed down, overlapping areas are multiplied"
|
||||
)
|
||||
|
||||
|
||||
class StairSchema(WithLineStringGeometrySchema, WithSpaceSchema, DjangoModelSchema):
|
||||
"""
|
||||
A line sharply dividing the accessible surface of a space into two different altitudes.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class RampSchema(WithPolygonGeometrySchema, WithSpaceSchema, DjangoModelSchema):
|
||||
"""
|
||||
An area in which the surface has an altitude gradient.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class BaseObstacleSchema(WithSpaceSchema, DjangoModelSchema):
|
||||
height: PositiveFloat = APIField(
|
||||
title="height",
|
||||
description="size of the obstacle in the z dimension"
|
||||
)
|
||||
altitude: NonNegativeFloat = APIField(
|
||||
title="altitude above ground",
|
||||
description="altitude above ground"
|
||||
)
|
||||
color: Optional[Color] = APIField(
|
||||
title="color",
|
||||
description="an optional color for this obstacle"
|
||||
)
|
||||
|
||||
|
||||
class ObstacleSchema(WithPolygonGeometrySchema, BaseObstacleSchema):
|
||||
"""
|
||||
An obstacle to be subtracted from the accessible surface of a space.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class LineObstacleSchema(WithLineStringGeometrySchema, BaseObstacleSchema):
|
||||
"""
|
||||
An obstacle to be subtracted from the accessible surface of a space, defined as a line with width.
|
||||
"""
|
||||
width: PositiveFloat = APIField(
|
||||
title="width",
|
||||
description="width of the line"
|
||||
)
|
||||
|
||||
|
||||
class ColumnSchema(WithPolygonGeometrySchema, WithSpaceSchema, DjangoModelSchema):
|
||||
"""
|
||||
A ceiling-high obstacle subtracted from the space, effectively creating a "building" again.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class POISchema(WithPointGeometrySchema, SpecificLocationSchema, WithSpaceSchema, DjangoModelSchema):
|
||||
"""
|
||||
A point of interest inside a space.
|
||||
|
||||
A POI is a specific location, and can therefor be routed to and from, as well as belong to location groups.
|
||||
"""
|
||||
pass
|
||||
|
|
|
@ -1,12 +1,10 @@
|
|||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from django.db import IntegrityError, transaction
|
||||
from ninja import Field as APIField
|
||||
from ninja import Router as APIRouter
|
||||
from ninja import Schema, UploadedFile
|
||||
from ninja.pagination import paginate
|
||||
from ninja.renderers import BaseRenderer
|
||||
from pydantic import PositiveInt, field_validator
|
||||
|
||||
from c3nav.api.exceptions import API404, APIConflict, APIRequestValidationFailed
|
||||
|
@ -38,8 +36,7 @@ class FirmwareBuildSchema(Schema):
|
|||
alias="binary",
|
||||
example="/media/firmware/012345/firmware.bin",
|
||||
description="download URL for the build binary",
|
||||
) # todo: downlaod differently?
|
||||
# todo: should not be none, but parse errors
|
||||
) # todo: downlaod differently?
|
||||
boards: set[BoardType] = APIField(
|
||||
description="set of boards that this build is compatible with",
|
||||
example={BoardType.C3NAV_LOCATION_PCB_REV_0_2.name, }
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue