40 lines
1,003 B
Python
40 lines
1,003 B
Python
# scopes.api.tracking
|
|
|
|
"""basic (generic) routes and views for tracks (records)"""
|
|
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends
|
|
from fastapi.responses import HTMLResponse
|
|
from pydantic import BaseModel, Field
|
|
import transaction
|
|
from typing import Annotated, Any, Self
|
|
|
|
from scopes.api import deps
|
|
from scopes.storage import tracking
|
|
from scopes.storage.common import Storage
|
|
|
|
class Track(BaseModel):
|
|
|
|
@classmethod
|
|
def create(cls, obj: Any) -> Self:
|
|
return cls(**obj.asDict())
|
|
|
|
uid: str = None
|
|
head: dict[str, str]
|
|
data: dict[str, Any]
|
|
timeStamp: datetime = None
|
|
|
|
# routes
|
|
|
|
router = APIRouter()
|
|
|
|
type Storage = Annotated[Storage, Depends(deps.storage)]
|
|
|
|
@router.get('/tracks', response_model=list[Track])
|
|
async def get_tracks(storage: Storage):
|
|
tracks = storage.create(tracking.Container)
|
|
result = [tr.asDict() for tr in tracks.query()]
|
|
#result = [Track.create(tr) for tr in tracks.query()]
|
|
transaction.commit()
|
|
return result
|
|
|