src/backend/lists.py

   1 #!/usr/bin/python3
   2
   3 import json
   4 import urllib.parse
   5
   6 from . import accounts
   7 from . import misc
   8
   9 class Lists(misc.Object):
  10         @property
  11         def url(self):
  12                 """
  13                         Returns the base URL of a Mailman instance
  14                 """
  15                 return self.settings.get("mailman-url")
  16
  17         @property
  18         def username(self):
  19                 return self.settings.get("mailman-username")
  20
  21         @property
  22         def password(self):
  23                 return self.settings.get("mailman-password")
  24
  25         async def _request(self, method, url, data=None):
  26                 headers, body = {}, None
  27
  28                 # URL
  29                 url = urllib.parse.urljoin(self.url, url)
  30
  31                 # For GET requests, append query arguments
  32                 if method == "GET":
  33                         if data:
  34                                 url = "%s?%s" % (url, urllib.parse.urlencode(data))
  35
  36                 # For POST/PUT encode all arguments as JSON
  37                 elif method in ("POST", "PUT", "PATCH"):
  38                         headers |= {
  39                                 "Content-Type" : "application/json",
  40                         }
  41
  42                         body = json.dumps(data)
  43
  44                 # Send the request and wait for a response
  45                 res = await self.backend.http_client.fetch(url, method=method,
  46                         headers=headers, body=body,
  47
  48                         # Authentication
  49                         auth_username=self.username, auth_password=self.password,
  50                 )
  51
  52                 # Decode JSON response
  53                 body = json.loads(res.body)
  54
  55                 # XXX handle errors
  56
  57                 return body
  58
  59         # Lists
  60
  61         async def _get_lists(self, *args, **kwargs):
  62                 lists = []
  63
  64                 # Fetch the response
  65                 response = await self._request(*args, **kwargs)
  66
  67                 # Fetch entries
  68                 for entry in response.get("entries", []):
  69                         list = List(self.backend, **entry)
  70                         lists.append(list)
  71
  72                 return lists
  73
  74         async def get_lists(self):
  75                 """
  76                         Fetches all available lists
  77                 """
  78                 data = {
  79                         "advertised" : True,
  80                 }
  81
  82                 return await self._get_lists("GET", "/api/3.1/lists", data=data)
  83
  84         async def get_subscribed_lists(self, account):
  85                 data = {
  86                         "subscriber" : account.email,
  87                         "role"       : "member",
  88                 }
  89
  90                 return await self._get_lists("GET", "/api/3.1/members/find", data=data)
  91
  92
  93 class List(misc.Object):
  94         def init(self, list_id, **kwargs):
  95                 self.list_id = list_id
  96
  97                 # Store all other data
  98                 self.data = kwargs
  99
 100         def __repr__(self):
 101                 return "<List %s>" % self.list_id
 102
 103         def __str__(self):
 104                 return self.display_name
 105
 106         def __eq__(self, other):
 107                 if isinstance(other, self.__class__):
 108                         return self.list_id == other.list_id
 109
 110                 return NotImplemented
 111
 112         def __lt__(self, other):
 113                 if isinstance(other, self.__class__):
 114                         return self.list_id < other.list_id
 115
 116                 return NotImplemented
 117
 118         def __len__(self):
 119                 return self.data.get("member_count")
 120
 121         @property
 122         def display_name(self):
 123                 return self.data.get("display_name")
 124
 125         @property
 126         def description(self):
 127                 return self.data.get("description")
 128
 129         @property
 130         def archive_url(self):
 131                 return "https://lists.ipfire.org/hyperkitty/list/%s/" % self.list_id
 132
 133         async def subscribe(self, account):
 134                 pass # XXX TODO
 135
 136         async def unsubscribe(self, account):
 137                 pass # XXX TODO