144 lines
3.9 KiB
Python
144 lines
3.9 KiB
Python
from ninja import Router, ModelSchema, Schema
|
|
from ninja.errors import HttpError
|
|
from config import models
|
|
from typing import List
|
|
from guardian.shortcuts import get_objects_for_user, assign_perm
|
|
from django.shortcuts import get_object_or_404
|
|
|
|
router = Router()
|
|
|
|
|
|
class Restream(ModelSchema):
|
|
class Meta:
|
|
model = models.Restream
|
|
fields = "__all__"
|
|
|
|
|
|
class RestreamPatch(ModelSchema):
|
|
class Meta:
|
|
model = models.Restream
|
|
exclude = ["id"]
|
|
fields_optional = "__all__"
|
|
extra = "forbid"
|
|
|
|
|
|
class Stream(ModelSchema):
|
|
class Meta:
|
|
model = models.Stream
|
|
fields = "__all__"
|
|
|
|
|
|
class StreamPatch(ModelSchema):
|
|
class Meta:
|
|
model = models.Stream
|
|
fields = ["name"]
|
|
fields_optional = "__all__"
|
|
extra = "forbid"
|
|
|
|
|
|
class StreamCreate(ModelSchema):
|
|
class Meta:
|
|
model = models.Stream
|
|
fields = ["name"]
|
|
extra = "forbid"
|
|
|
|
|
|
@router.get('/streams', response=List[Stream])
|
|
def list_streams(request):
|
|
return get_objects_for_user(request.user, 'view_stream', models.Stream.objects.all())
|
|
|
|
|
|
@router.post('/streams', response=Stream)
|
|
def create_stream(request, payload: StreamCreate):
|
|
stream = models.Stream.objects.create(**payload.dict())
|
|
assign_perm('view_stream', request.user, stream)
|
|
assign_perm('change_stream', request.user, stream)
|
|
assign_perm('delete_stream', request.user, stream)
|
|
return stream
|
|
|
|
|
|
@router.get('/streams/{id}', response=Stream)
|
|
def get_stream(request, id: int):
|
|
stream = get_object_or_404(models.Stream, id=id)
|
|
|
|
if not request.user.has_perm('view_stream', stream):
|
|
raise HttpError(401, "unauthorized")
|
|
return stream
|
|
|
|
|
|
@router.patch('/streams/{id}', response=Stream)
|
|
def update_stream(request, id: int, payload: StreamPatch):
|
|
stream = get_object_or_404(models.Stream, id=id)
|
|
|
|
if not request.user.has_perm('change_stream', stream):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
stream.name = payload.name
|
|
stream.save()
|
|
return stream
|
|
|
|
|
|
@router.delete('/streams/{id}', response=None)
|
|
def delete_stream(request, id: int):
|
|
stream = get_object_or_404(models.Stream, id=id)
|
|
|
|
if not request.user.has_perm('delete_stream', stream):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
stream.delete()
|
|
|
|
|
|
@router.get('/restreams', response=List[Restream])
|
|
def list_restreams(request):
|
|
return get_objects_for_user(request.user, 'view_restream', models.Restream.objects.all())
|
|
|
|
|
|
@router.post('/restreams', response=Restream)
|
|
def create_restream(request, payload: Restream):
|
|
if not request.user.has_perm('view_stream', payload.stream):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
restream = models.Restream.objects.create(**payload.dict())
|
|
assign_perm('view_restream', request.user, restream)
|
|
assign_perm('change_restream', request.user, restream)
|
|
assign_perm('delete_restream', request.user, restream)
|
|
return restream
|
|
|
|
|
|
@router.get('/restreams/{id}', response=Restream)
|
|
def get_restream(request, id: int):
|
|
restream = get_object_or_404(models.Restream, id=id)
|
|
|
|
if not request.user.has_perm('view_restream', restream):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
return restream
|
|
|
|
|
|
@router.patch('/restreams/{id}', response=Restream)
|
|
def update_restream(request, id: int, payload: RestreamPatch):
|
|
restream = get_object_or_404(models.Restream, id=id)
|
|
|
|
if not request.user.has_perm('change_restream', restream):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
if payload.stream:
|
|
payload.stream = get_object_or_404(models.Stream, id=payload.stream)
|
|
|
|
if not request.user.has_perm('view_stream', payload.stream):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
for key, value in payload.dict(exclude_unset=True).items():
|
|
setattr(restream, key, value)
|
|
restream.save()
|
|
return restream
|
|
|
|
|
|
@router.delete('/restreams/{id}', response=None)
|
|
def delete_restream(request, id: int):
|
|
restream = get_object_or_404(models.Restream, id=id)
|
|
|
|
if not request.user.has_perm('delete_restream', restream):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
restream.delete() |