fix missing annotations rpc methods (84f1e885) · Commits · certic / Jama

annotations/rpc/init.py

deleted100644 → 0

+0 −0

Empty file deleted.

annotations/rpc/methods.py

deleted100644 → 0

+0 −102

Original line number	Diff line number	Diff line
		from rpc.methods import (
		_rpc_groups,
		ServiceException,
		_check_project_permission,
		_log_call,
		)
		from django.contrib.auth.models import User
		from resources.models import Resource
		from annotations.models import Annotation
		from typing import List
		from rpc.const import * # noqa: F403


		def _serialize_annotation(annotation: Annotation) -> dict:
		return {
		"id": annotation.id,
		"owner": annotation.owner.username,
		"data": annotation.data,
		"created_at": annotation.created_at.isoformat(),
		"updated_at": annotation.updated_at.isoformat(),
		"resource_id": annotation.resource_id,
		}


		def _fetch_resource(resource_id: int) -> Resource:
		resource_instance = Resource.objects.filter(
		pk=resource_id, deleted_at__isnull=True
		).first()
		if not resource_instance:
		raise ServiceException(NO_SUCH_RESOURCE) # noqa: F405
		return resource_instance


		@_rpc_groups(["Annotations"])
		def list_annotations(user: User, resource_id: int) -> List[dict]:
		"""
		List all annotations for a given resource
		"""
		resource_instance = _fetch_resource(resource_id)
		_check_project_permission(user, resource_instance.ptr_project, PERM_RESOURCE_READ) # noqa: F405
		annotations_list = []
		for annotation in Annotation.objects.filter(resource=resource_instance).order_by(
		"created_at"
		):
		annotations_list.append(_serialize_annotation(annotation))
		return annotations_list


		@_rpc_groups(["Annotations"])
		@_log_call
		def create_annotation(user: User, resource_id: int, data: dict) -> dict:
		"""
		Adds an annotation, returning the serialized annotation:

		```
		{
		"id": annotation.id,
		"owner": annotation.owner.username,
		"data": annotation.data,
		"created_at": annotation.created_at.isoformat(),
		"updated_at": annotation.updated_at.isoformat(),
		"resource_id": annotation.resource_id,
		}
		```
		"""
		resource_instance = _fetch_resource(resource_id)
		_check_project_permission(user, resource_instance.ptr_project, PERM_RESOURCE_UPDATE) # noqa: F405
		annotation = Annotation.objects.create(
		resource=resource_instance, owner=user, data=data
		)
		return _serialize_annotation(annotation)


		@_rpc_groups(["Annotations"])
		@_log_call
		def update_annotation(user: User, annotation_id: int, data: dict) -> dict:
		"""
		Updates an annotation, returning the serialized annotation
		"""
		annotation_instance = Annotation.objects.filter(
		pk=annotation_id, owner=user
		).first()
		if not annotation_instance:
		raise ServiceException("No such annotation")
		annotation_instance.data = data
		annotation_instance.save()
		return _serialize_annotation(annotation_instance)


		@_rpc_groups(["Annotations"])
		@_log_call
		def delete_annotation(user: User, annotation_id: int) -> bool:
		"""
		Deletes an annotation
		"""
		annotation_instance = Annotation.objects.filter(
		pk=annotation_id, owner=user
		).first()
		if not annotation_instance:
		raise ServiceException("No such annotation")
		annotation_instance.delete()
		return True

rpc/methods.py

+93 −0

Original line number	Diff line number	Diff line
		@@ -43,6 +43,8 @@ from resources.models import (
		XLSX_MULTIPLE_VALUES_SEPARATOR,
		)

		from annotations.models import Annotation

		from rpc import serializers
		from rpc.cache import SerializerCache
		from rpc.const import * # noqa: F403
		@@ -3323,3 +3325,94 @@ def newstuff(user: User, something: int) -> List[dict]:
		q = Resource.objects.filter(title="toto")
		for r in q.all():
		print(r.title)


		def _serialize_annotation(annotation: Annotation) -> dict:
		return {
		"id": annotation.id,
		"owner": annotation.owner.username,
		"data": annotation.data,
		"created_at": annotation.created_at.isoformat(),
		"updated_at": annotation.updated_at.isoformat(),
		"resource_id": annotation.resource_id,
		}


		def _fetch_resource(resource_id: int) -> Resource:
		resource_instance = Resource.objects.filter(
		pk=resource_id, deleted_at__isnull=True
		).first()
		if not resource_instance:
		raise ServiceException(NO_SUCH_RESOURCE) # noqa: F405
		return resource_instance


		@_rpc_groups(["Annotations"])
		def list_annotations(user: User, resource_id: int) -> List[dict]:
		"""
		List all annotations for a given resource
		"""
		resource_instance = _fetch_resource(resource_id)
		_check_project_permission(user, resource_instance.ptr_project, PERM_RESOURCE_READ) # noqa: F405
		annotations_list = []
		for annotation in Annotation.objects.filter(resource=resource_instance).order_by(
		"created_at"
		):
		annotations_list.append(_serialize_annotation(annotation))
		return annotations_list


		@_rpc_groups(["Annotations"])
		@_log_call
		def create_annotation(user: User, resource_id: int, data: dict) -> dict:
		"""
		Adds an annotation, returning the serialized annotation:

		```
		{
		"id": annotation.id,
		"owner": annotation.owner.username,
		"data": annotation.data,
		"created_at": annotation.created_at.isoformat(),
		"updated_at": annotation.updated_at.isoformat(),
		"resource_id": annotation.resource_id,
		}
		```
		"""
		resource_instance = _fetch_resource(resource_id)
		_check_project_permission(user, resource_instance.ptr_project, PERM_RESOURCE_UPDATE) # noqa: F405
		annotation = Annotation.objects.create(
		resource=resource_instance, owner=user, data=data
		)
		return _serialize_annotation(annotation)


		@_rpc_groups(["Annotations"])
		@_log_call
		def update_annotation(user: User, annotation_id: int, data: dict) -> dict:
		"""
		Updates an annotation, returning the serialized annotation
		"""
		annotation_instance = Annotation.objects.filter(
		pk=annotation_id, owner=user
		).first()
		if not annotation_instance:
		raise ServiceException("No such annotation")
		annotation_instance.data = data
		annotation_instance.save()
		return _serialize_annotation(annotation_instance)


		@_rpc_groups(["Annotations"])
		@_log_call
		def delete_annotation(user: User, annotation_id: int) -> bool:
		"""
		Deletes an annotation
		"""
		annotation_instance = Annotation.objects.filter(
		pk=annotation_id, owner=user
		).first()
		if not annotation_instance:
		raise ServiceException("No such annotation")
		annotation_instance.delete()
		return True