lists: Implement some basic code to talk to Mailman

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

#!/usr/bin/python3

import json
import urllib.parse

from . import accounts
from . import misc

class Lists(misc.Object):
	@property
	def url(self):
		"""
			Returns the base URL of a Mailman instance
		"""
		return self.settings.get("mailman-url")

	@property
	def username(self):
		return self.settings.get("mailman-username")

	@property
	def password(self):
		return self.settings.get("mailman-password")

	async def _request(self, method, url, data=None):
		headers, body = {}, None

		# URL
		url = urllib.parse.urljoin(self.url, url)

		# 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", "PATCH"):
			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,

			# Authentication
			auth_username=self.username, auth_password=self.password,
		)

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

		# XXX handle errors

		return body

	async def _get_lists(self, *args, **kwargs):
		lists = []

		# Fetch the response
		response = await self._request(*args, **kwargs)

		# Fetch entries
		for entry in response.get("entries", []):
			list = List(self.backend, **entry)
			lists.append(list)

		return lists

	async def get_lists(self):
		"""
			Fetches all available lists
		"""
		data = {
			"advertised" : True,
		}

		return await self._get_lists("GET", "/3.1/lists", data=data)

	async def get_subscribed_lists(self, account):
		data = {
			"subscriber" : account.email,
			"role"       : "member",
		}

		return await self._get_lists("GET", "/3.1/members/find", data=data)


class List(misc.Object):
	def init(self, list_id, **kwargs):
		self.list_id = list_id

		# Store all other data
		self.data = kwargs

	def __repr__(self):
		return "<List %s>" % self.list_id

	def __str__(self):
		return self.display_name

	def __eq__(self, other):
		if isinstance(other, self.__class__):
			return self.list_id == other.list_id

		return NotImplemented

	def __lt__(self, other):
		if isinstance(other, self.__class__):
			return self.list_id < other.list_id

		return NotImplemented

	def __len__(self):
		return self.data.get("member_count")

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

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

	async def subscribe(self, account):
		pass # XXX TODO

	async def unsubscribe(self, account):
		pass # XXX TODO