90 lines
2.2 KiB
Python
90 lines
2.2 KiB
Python
from ninja import NinjaAPI
|
|
from ninja import Schema
|
|
from ninja.security import APIKeyQuery
|
|
import datetime
|
|
from ..accounts.models import User
|
|
from .models import ThingEdit, Thing as ThingModel, View as ViewModel
|
|
|
|
|
|
class ApiKey(APIKeyQuery):
|
|
param_name = "api_key"
|
|
|
|
def authenticate(self, request, key):
|
|
try:
|
|
# TODO: replace with secret keys
|
|
user = User.objects.get(username=key)
|
|
return user
|
|
except User.DoesNotExist:
|
|
pass
|
|
|
|
|
|
api = NinjaAPI(auth=ApiKey())
|
|
|
|
|
|
class View(Schema):
|
|
name: str
|
|
columns: dict
|
|
filters: dict
|
|
sort: str
|
|
|
|
|
|
class Thing(Schema):
|
|
id: int
|
|
data: dict
|
|
type: str
|
|
|
|
|
|
@api.post("/view/")
|
|
def sync_view(request, view: View):
|
|
ViewModel.objects.update_or_create(
|
|
name=view.name,
|
|
defaults=dict(
|
|
columns=view.columns,
|
|
filters=view.filters,
|
|
sort=view.sort,
|
|
),
|
|
)
|
|
|
|
|
|
@api.post("/thing/")
|
|
def sync_thing(request, thing: Thing):
|
|
action = "none"
|
|
try:
|
|
old_thing = ThingModel.objects.get(user=request.auth, thing_id=thing.id)
|
|
print("update", thing)
|
|
if old_thing.data != thing.data:
|
|
ThingEdit.objects.create(
|
|
thing=old_thing,
|
|
data=old_thing.data,
|
|
timestamp=old_thing.synced_at,
|
|
)
|
|
old_thing.data = thing.data
|
|
old_thing.synced_at = datetime.datetime.utcnow()
|
|
old_thing.save()
|
|
thing = old_thing
|
|
action = "updated"
|
|
except ThingModel.DoesNotExist:
|
|
print("new", thing)
|
|
thing = ThingModel.objects.create(
|
|
user=request.auth,
|
|
thing_id=thing.id,
|
|
data=thing.data,
|
|
type=thing.type,
|
|
synced_at=datetime.datetime.utcnow(),
|
|
)
|
|
action = "created"
|
|
# TODO: deleted
|
|
return {"action": action}
|
|
|
|
|
|
@api.get("/thing/{id}")
|
|
def get_thing(request, id: int, key: str):
|
|
thing = ThingModel.objects.get(user=request.auth, thing_id=id)
|
|
return thing.data
|
|
|
|
|
|
@api.get("/thing/latest_changes")
|
|
def latest_changes(request, since: str):
|
|
updates = ThingModel.objects.filter(user=request.auth, updated_at__gte=since)
|
|
return [{"id": u.id, "updated_at": u.updated_at} for u in updates]
|