src/mailman/rest/queues.py

   1 # Copyright (C) 2015-2023 by the Free Software Foundation, Inc.
   2 #
   3 # This file is part of GNU Mailman.
   4 #
   5 # GNU Mailman is free software: you can redistribute it and/or modify it under
   6 # the terms of the GNU General Public License as published by the Free
   7 # Software Foundation, either version 3 of the License, or (at your option)
   8 # any later version.
   9 #
  10 # GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
  11 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  13 # more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along with
  16 # GNU Mailman.  If not, see <https://www.gnu.org/licenses/>.
  17
  18 """<api>/queues."""
  19
  20 from mailman.app.inject import inject_text
  21 from mailman.config import config
  22 from mailman.interfaces.listmanager import IListManager
  23 from mailman.rest.helpers import (
  24     bad_request,
  25     CollectionMixin,
  26     created,
  27     etag,
  28     no_content,
  29     not_found,
  30     okay,
  31 )
  32 from mailman.rest.validator import Validator
  33 from public import public
  34 from zope.component import getUtility
  35
  36
  37 class _QueuesBase(CollectionMixin):
  38     """Shared base class for queues."""
  39
  40     def _resource_as_dict(self, name):
  41         """See `CollectionMixin`."""
  42         switchboard = config.switchboards[name]
  43         files = switchboard.files
  44         return dict(
  45             name=switchboard.name,
  46             directory=switchboard.queue_directory,
  47             count=len(files),
  48             files=files,
  49             self_link=self.api.path_to('queues/{}'.format(name)),
  50             )
  51
  52     def _get_collection(self, request):
  53         """See `CollectionMixin`."""
  54         return sorted(config.switchboards)
  55
  56
  57 @public
  58 class AQueue(_QueuesBase):
  59     """A single queue."""
  60
  61     def __init__(self, name):
  62         self._name = name
  63
  64     def on_get(self, request, response):
  65         """Return a single queue resource."""
  66         if self._name not in config.switchboards:
  67             not_found(response)
  68         else:
  69             okay(response, self._resource_as_json(self._name))
  70
  71     def on_post(self, request, response):
  72         """Inject a message into the queue."""
  73         try:
  74             validator = Validator(list_id=str,
  75                                   text=str)
  76             values = validator(request)
  77         except ValueError as error:
  78             bad_request(response, str(error))
  79             return
  80         list_id = values['list_id']
  81         mlist = getUtility(IListManager).get_by_list_id(list_id)
  82         if mlist is None:
  83             bad_request(response, 'No such list: {}'.format(list_id))
  84             return
  85         try:
  86             filebase = inject_text(
  87                 mlist, values['text'], switchboard=self._name)
  88         except Exception as error:
  89             bad_request(response, str(error))
  90             return
  91         else:
  92             location = self.api.path_to(
  93                 'queues/{}/{}'.format(self._name, filebase))
  94             created(response, location)
  95
  96
  97 @public
  98 class AQueueFile:
  99     def __init__(self, name, filebase):
 100         self._name = name
 101         self._filebase = filebase
 102
 103     def on_delete(self, request, response):
 104         """Delete the queue file."""
 105         switchboard = config.switchboards.get(self._name)
 106         if switchboard is None:
 107             not_found(response, 'No such queue: {}'.format(self._name))
 108             return
 109         try:
 110             switchboard.dequeue(self._filebase)
 111         except FileNotFoundError:
 112             not_found(response,
 113                       'No such queue file: {}'.format(self._filebase))
 114         else:
 115             no_content(response)
 116
 117
 118 @public
 119 class AllQueues(_QueuesBase):
 120     """All queues."""
 121
 122     def on_get(self, request, response):
 123         """<api>/queues"""
 124         resource = self._make_collection(request)
 125         resource['self_link'] = self.api.path_to('queues')
 126         okay(response, etag(resource))