2025-summer/team-3
12
Fork 0
Code Issues Pull requests Projects Releases Activity

update django-ninja, including pydantic v2 and add provisional level api

Browse source
This commit is contained in:
Laura Klünder 2023-11-18 21:29:35 +01:00
parent f89d069ab1
commit b2aa76ba2d
20 changed files with 510 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

61
src/c3nav/mesh/newapi.py
View file

@ -1,44 +1,43 @@
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 pydantic import validator
from pydantic import PositiveInt, field_validator
from c3nav.api.exceptions import APIConflict, APIRequestValidationFailed, API404
from c3nav.api.newauth import BearerAuth, auth_permission_responses, auth_responses
from c3nav.mesh.dataformats import BoardType, FirmwareImage, ChipType
from c3nav.mesh.models import FirmwareVersion, FirmwareBuild
from c3nav.api.exceptions import API404, APIConflict, APIRequestValidationFailed
from c3nav.api.newauth import BearerAuth, auth_permission_responses, auth_responses, validate_responses
from c3nav.mesh.dataformats import BoardType, ChipType, FirmwareImage
from c3nav.mesh.models import FirmwareBuild, FirmwareVersion
api_router = APIRouter(tags=["mesh"])
mesh_api_router = APIRouter(tags=["mesh"])
class FirmwareBuildSchema(Schema):
id: int
id: PositiveInt
variant: str = APIField(..., example="c3uart")
chip: ChipType = APIField(..., example=ChipType.ESP32_C3.name)
sha256_hash: str = APIField(..., regex=r"^[0-9a-f]{64}$")
url: str = APIField(..., alias="binary", example="/media/firmware/012345/firmware.bin")
sha256_hash: str = APIField(..., pattern=r"^[0-9a-f]{64}$")
url: str = APIField(..., alias="binary", example="/media/firmware/012345/firmware.bin") # todo: downlaod differently?
# todo: should not be none, but parse errors
boards: list[BoardType] = APIField(None, example=[BoardType.C3NAV_LOCATION_PCB_REV_0_2.name, ])
@staticmethod
def resolve_chip(obj):
# todo: do this in model? idk
return ChipType(obj.chip)
class Config(Schema.Config):
pass
class FirmwareSchema(Schema):
id: int
id: PositiveInt
project_name: str = APIField(..., example="c3nav_positioning")
version: str = APIField(..., example="499837d-dirty")
idf_version: str = APIField(..., example="v5.1-476-g3187b8b326")
created: datetime
builds: list[FirmwareBuildSchema]
@validator('builds')
@field_validator('builds')
def builds_variants_must_be_unique(cls, builds):
if len(set(build.variant for build in builds)) != len(builds):
raise ValueError("builds must have unique variant identifiers")
@ -49,15 +48,15 @@ class Error(Schema):
detail: str
@api_router.get('/firmwares/', summary="List available firmwares",
response={200: list[FirmwareSchema], **auth_responses})
@mesh_api_router.get('/firmwares/', summary="List available firmwares",
response={200: list[FirmwareSchema], **validate_responses, **auth_responses})
@paginate
def firmware_list(request):
return FirmwareVersion.objects.all()
@api_router.get('/firmwares/{firmware_id}/', summary="Get specific firmware",
response={200: FirmwareSchema, **API404.dict(), **auth_responses})
@mesh_api_router.get('/firmwares/{firmware_id}/', summary="Get specific firmware",
response={200: FirmwareSchema, **API404.dict(), **auth_responses})
def firmware_detail(request, firmware_id: int):
try:
return FirmwareVersion.objects.get(id=firmware_id)
@ -65,9 +64,9 @@ def firmware_detail(request, firmware_id: int):
raise API404("Firmware not found")
@api_router.get('/firmwares/{firmware_id}/{variant}/image_data',
summary="Get header data of firmware build image",
response={200: FirmwareImage.schema, **API404.dict(), **auth_responses})
@mesh_api_router.get('/firmwares/{firmware_id}/{variant}/image_data',
summary="Get header data of firmware build image",
response={200: FirmwareImage.schema, **API404.dict(), **auth_responses})
def firmware_build_image(request, firmware_id: int, variant: str):
try:
build = FirmwareBuild.objects.get(version_id=firmware_id, variant=variant)
@ -76,9 +75,9 @@ def firmware_build_image(request, firmware_id: int, variant: str):
raise API404("Firmware or firmware build not found")
@api_router.get('/firmwares/{firmware_id}/{variant}/project_description',
summary="Get project description of firmware build",
response={200: dict, **API404.dict(), **auth_responses})
@mesh_api_router.get('/firmwares/{firmware_id}/{variant}/project_description',
summary="Get project description of firmware build",
response={200: dict, **API404.dict(), **auth_responses})
def firmware_project_description(request, firmware_id: int, variant: str):
try:
return FirmwareBuild.objects.get(version_id=firmware_id, variant=variant).firmware_description
@ -97,18 +96,20 @@ class UploadFirmwareSchema(Schema):
project_name: str = APIField(..., example="c3nav_positioning")
version: str = APIField(..., example="499837d-dirty")
idf_version: str = APIField(..., example="v5.1-476-g3187b8b326")
builds: list[UploadFirmwareBuildSchema] = APIField(..., min_items=1, unique_items=True)
builds: list[UploadFirmwareBuildSchema] = APIField(..., min_items=1)
@validator('builds')
@field_validator('builds')
def builds_variants_must_be_unique(cls, builds):
if len(set(build.variant for build in builds)) != len(builds):
raise ValueError("builds must have unique variant identifiers")
return builds
@api_router.post('/firmwares/upload', summary="Upload firmware", auth=BearerAuth(superuser=True),
description="your OpenAPI viewer might not show it: firmware_data is UploadFirmwareSchema as json",
response={200: FirmwareSchema, **auth_permission_responses, **APIConflict.dict()})
@mesh_api_router.post(
'/firmwares/upload', summary="Upload firmware", auth=BearerAuth(superuser=True),
description="your OpenAPI viewer might not show it: firmware_data is UploadFirmwareSchema as json",
response={200: FirmwareSchema, **validate_responses, **auth_permission_responses, **APIConflict.dict()}
)
def firmware_upload(request, firmware_data: UploadFirmwareSchema, binary_files: list[UploadedFile]):
binary_files_by_name = {binary_file.name: binary_file for binary_file in binary_files}
if len([binary_file.name for binary_file in binary_files]) != len(binary_files_by_name):