78 lines
2.0 KiB
Python
78 lines
2.0 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 Pull(ModelSchema):
|
|
class Meta:
|
|
model = models.Pull
|
|
fields = "__all__"
|
|
|
|
|
|
class PullPatch(ModelSchema):
|
|
class Meta:
|
|
model = models.Pull
|
|
exclude = ["id"]
|
|
fields_optional = "__all__"
|
|
extra = "forbid"
|
|
|
|
|
|
@router.get('/', response=List[Pull])
|
|
def list_pulls(request):
|
|
return get_objects_for_user(request.user, 'view_pull', models.Pull.objects.all())
|
|
|
|
|
|
@router.post('/', response=Pull)
|
|
def create_pull(request, payload: Pull):
|
|
if not request.user.has_perm('view_stream', payload.stream):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
pull = models.Pull.objects.create(**payload.dict())
|
|
assign_perm( 'view_pull', request.user, pull)
|
|
assign_perm('change_pull', request.user, pull)
|
|
assign_perm('delete_pull', request.user, pull)
|
|
return pull
|
|
|
|
|
|
@router.get('/{id}', response=Pull)
|
|
def get_pull(request, id: int):
|
|
pull = get_object_or_404(models.Pull, id=id)
|
|
|
|
if not request.user.has_perm('view_pull', pull):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
return pull
|
|
|
|
|
|
@router.patch('/{id}', response=Pull)
|
|
def patch_pull(request, id: int, payload: PullPatch):
|
|
pull = get_object_or_404(models.Pull, id=id)
|
|
|
|
if not request.user.has_perm('change_pull', pull):
|
|
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(pull, key, value)
|
|
pull.save()
|
|
return pull
|
|
|
|
|
|
@router.delete('/{id}', response=None)
|
|
def delete_pull(request, id: int):
|
|
pull = get_object_or_404(models.Pull, id=id)
|
|
|
|
if not request.user.has_perm('delete_pull', pull):
|
|
raise HttpError(401, "unauthorized")
|
|
|
|
pull.delete()
|