black (3bd9fa53) · Commits · certic / Jama

rpc/methods.py

+42 −11

Original line number	Diff line number	Diff line
		@@ -32,10 +32,21 @@ SUPERUSER_NEEDED = "only for superusers"


		class ServiceException(Exception):
		"""
		Base exception for all RPC methods.
		Accepts a message
		"""

		pass


		def _require_superuser(func):
		"""
		(Decorator)

		Force superuser check before function execution
		"""

		@_wraps(func)
		def wrapper(args, *kwargs):
		user = args[0] # user MUST be first argument
		@@ -46,7 +57,14 @@ def _require_superuser(func):
		return wrapper


		def _rpc_groups(groups: list):
		def _rpc_groups(groups: List[str]):
		"""
		(Decorator)

		Categorize the function by adding a rpc_groups
		property to it.
		"""

		def decorator(func):
		@_wraps(func)
		def wrapper(args, *kwargs):
		@@ -1545,7 +1563,7 @@ def set_tag(


		@_rpc_groups(["Tags"])
		def remove_tag(user: User, uid: str) -> bool:
		def delete_tag(user: User, uid: str) -> bool:
		"""
		Remove (delete) a tag based on its uid.

		@@ -1809,7 +1827,7 @@ def activate_rpc_access(user: User, user_name: str, api_key: str) -> bool:
		Add access to the RPC API for the given user name with the given API key.
		A new user will be created if none is available with given user name.

		Note: only user with admin privileges can use this.
		Requires superuser.
		"""
		user_instance, _ = User.objects.get_or_create(username=user_name)
		key_instance, _ = APIKey.objects.get_or_create(user=user_instance, key=api_key)
		@@ -1825,7 +1843,7 @@ def deactivate_rpc_access(user: User, user_name: str, api_key: str) -> bool:
		Deactivate access to the RPC API for the given user name and API key.
		Only the access (API key) is removed, not the user.

		Note: only user with admin privileges can use this.
		Requires superuser.
		"""
		try:
		user_instance = User.objects.get(username=user_name)
		@@ -1840,6 +1858,11 @@ def deactivate_rpc_access(user: User, user_name: str, api_key: str) -> bool:
		@_rpc_groups(["Access"])
		@_require_superuser
		def create_project(user: User, project_label: str, project_description: str) -> dict:
		"""
		Create a new project.

		Requires superuser.
		"""
		project = Project.objects.create(
		label=project_label, description=project_description
		)
		@@ -1934,12 +1957,18 @@ def list_roles(user: User, project_id: int) -> List[Dict]:

		@_rpc_groups(["Access"])
		@_require_superuser
		def set_role(user: User, project_id: int, label: str, permissions: List[int]) -> Dict:
		def set_role(
		user: User, project_id: int, role_label: str, permissions: List[int]
		) -> Dict:
		"""
		Create or update a role on a project, with the given permissions.

		Requires superuser.
		"""
		project = _check_project_permission(user, project_id, "permission.write")
		role, created = Role.objects.get_or_create(label=label.strip(), project=project)
		role, created = Role.objects.get_or_create(
		label=role_label.strip(), project=project
		)
		role.permissions.clear()
		for perm_id in permissions:
		try:
		@@ -1952,9 +1981,11 @@ def set_role(user: User, project_id: int, label: str, permissions: List[int]) ->

		@_require_superuser
		@_rpc_groups(["Access"])
		def delete_role(user: User, role_id: int) -> bool:
		role = Role.objects.filter(pk=role_id).first()
		if not role:
		return False
		role.delete()
		def delete_role(user: User, project_id: int, role_label: str) -> bool:
		"""
		Delete role within given project.

		Requires superuser
		"""
		Role.objects.filter(label=role_label, project_id=project_id).delete()
		return True