lists: Implement some basic code to talk to Mailman

[ipfire.org.git] / src / backend / bugzilla.py

#!/usr/bin/python3

import json
import urllib.parse

from . import httpclient
from . import misc
from .decorators import *

class BugzillaError(Exception):
	pass

class Bugzilla(misc.Object):
	def init(self, api_key=None):
		if api_key is None:
			api_key = self.settings.get("bugzilla-api-key")

		# Store the API key
		self.api_key = api_key

	@property
	def url(self):
		"""
			Returns the base URL of a Bugzilla instance
		"""
		return self.settings.get("bugzilla-url")

	def make_url(self, *args, **kwargs):
		"""
			Composes a URL based on the base URL
		"""
		url = urllib.parse.urljoin(self.url, *args)

		# Append any query arguments
		if kwargs:
			url = "%s?%s" % (url, urllib.parse.urlencode(kwargs))

		return url

	async def _request(self, method, url, data=None):
		if data is None:
			data = {}

		# Headers
		headers = {
			# Authenticate all requests
			"X-BUGZILLA-API-KEY" : self.api_key,
		}

		# Make the URL
		url = self.make_url(url)

		# Fallback authentication because some API endpoints
		# do not accept the API key in the header
		data |= { "api_key" : self.api_key }

		# Encode body
		body = None

		# For GET requests, append query arguments
		if method == "GET":
			if data:
				url = "%s?%s" % (url, urllib.parse.urlencode(data))

		# For POST/PUT encode all arguments as JSON
		elif method in ("POST", "PUT"):
			headers |= {
				"Content-Type" : "application/json",
			}

			body = json.dumps(data)

		# Send the request and wait for a response
		res = await self.backend.http_client.fetch(
			url, method=method, headers=headers, body=body)

		# Decode JSON response
		body = json.loads(res.body)

		# Check for any errors
		if "error" in body:
			# Fetch code and message
			code, message = body.get("code"), body.get("message")

			# Handle any so far unhandled errors
			raise BugzillaError(message)

		# Return an empty response
		return body

	async def get_user(self, uid):
		"""
			Fetches a user from Bugzilla
		"""
		try:
			response = await self._request("GET", "/rest/user/%s" % uid)

		# Return nothing if the user could not be found
		except httpclient.HTTPError as e:
			if e.code == 404:
				return

			raise e

		# Return the user object
		for data in response.get("users"):
			return User(self.backend, data)



class User(misc.Object):
	def init(self, data):
		self.data = data

	@property
	def id(self):
		return self.data.get("id")

	async def _update(self, **kwargs):
		# Send the request
		await self.backend.bugzilla._request("PUT", "/rest/user/%s" % self.id, **kwargs)

		# XXX apply changes to the User object?

	async def disable(self, text=None):
		"""
			Disables this user
		"""
		if not text:
			text = "DISABLED"

		# Update the user
		await self._update(data={
			"login_denied_text" : text,
		})