36 lines
988 B
Python
36 lines
988 B
Python
# scopes.api.tracking
|
|
|
|
"""basic (generic) routes and views for tracks (records)"""
|
|
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Body
|
|
from pydantic import BaseModel, Field
|
|
from typing import Annotated, Any, Self
|
|
|
|
from scopes.api import deps
|
|
from scopes.storage import tracking
|
|
|
|
class Track(BaseModel):
|
|
uid: str | None = None
|
|
head: dict[str, str]
|
|
data: dict[str, Any]
|
|
timeStamp: datetime | None = None
|
|
|
|
# routes
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get('/tracks', response_model=list[Track])
|
|
async def get_tracks(storage: deps.Storage):
|
|
tracks = storage.create(tracking.Container)
|
|
result = [tr.asDict() for tr in tracks.query()]
|
|
return result
|
|
|
|
@router.post('/tracks', response_model=Track)
|
|
async def post_track(track: Annotated[Track, Body()], storage: deps.Storage):
|
|
tracks = storage.create(tracking.Container)
|
|
hd = track.head
|
|
tr = tracking.Track(hd['taskId'], hd['userName'], data=track.data)
|
|
tracks.save(tr)
|
|
return tr.asDict()
|
|
|