Merge pull request #164 from slimta/pytype

Consolidate plugin repos into main

.gitignore

@@ -12,3 +12,4 @@ examples/cert.pem
 *.pyc
 *.pyo
 .*.swp
+.pytype/

.lvimrc

@@ -0,0 +1,4 @@
+let g:ale_fix_on_save = 1
+let g:ale_fixers = {
+\   'python': ['autopep8'],
+\}

.travis.yml

@@ -1,14 +1,18 @@
 language: python
 python:
-  - "2.7"
-  - "3.5"
-  - "3.6"
-  - "3.7"
+  - "2.7.18"
+  - "3.5.9"
+  - "3.6.12"
+  - "3.7.9"
+  - "3.8.6"
 dist: xenial     # https://github.com/travis-ci/travis-ci/issues/9069#issuecomment-425720905
 install:
   - travis_retry pip install -r test/requirements.txt
   - travis_retry pip install coveralls
   - travis_retry pip install -e .
-script: py.test --cov=slimta
+script:
+  - py.test --cov=slimta
+  - flake8 slimta
+  - pytype -k
 after_success:
   - coveralls

setup.cfg

@@ -13,3 +13,5 @@ exclude_lines =
   pragma: no cover
   raise NotImplementedError
 
+[pytype]
+inputs = slimta

setup.py

@@ -23,7 +23,7 @@
 
 
 setup(name='python-slimta',
-      version='4.0.11',
+      version='4.1.0',
       author='Ian Good',
       author_email='icgood@gmail.com',
       description='Lightweight, asynchronous SMTP libraries.',
@@ -35,16 +35,23 @@
                         'pysasl >= 0.4.0, < 0.5',
                         'pycares < 3.0.0; python_version < "3.0"',
                         'pycares >= 1; python_version >= "3.0"'],
+      extras_require={'spf': ['pyspf', 'py3dns; python_version >= "3.0"',
+                              'pydns; python_version < "3.0"',
+                              'ipaddr; python_version < "3.0"'],
+                      'redis': ['redis'],
+                      'aws': ['boto'],
+                      'disk': ['pyaio >= 0.4; platform_system == "Linux"']},
       classifiers=['Development Status :: 3 - Alpha',
                    'Topic :: Communications :: Email :: Mail Transport Agents',
                    'Intended Audience :: Developers',
                    'Intended Audience :: Information Technology',
                    'License :: OSI Approved :: MIT License',
                    'Programming Language :: Python',
                    'Programming Language :: Python :: 2.7',
-                   'Programming Language :: Python :: 3.4',
                    'Programming Language :: Python :: 3.5',
-                   'Programming Language :: Python :: 3.6'])
+                   'Programming Language :: Python :: 3.6',
+                   'Programming Language :: Python :: 3.7',
+                   'Programming Language :: Python :: 3.8'])
 
 
 # vim:et:fdm=marker:sts=4:sw=4:ts=4

slimta/cloudstorage/__init__.py

@@ -0,0 +1,127 @@
+# Copyright (c) 2013 Ian C. Good
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+#
+
+"""Package containing a module for the different cloud service providers along
+with any necessary helper modules.
+
+.. _Cloud Files: http://www.rackspace.com/cloud/files/
+.. _Cloud Queues: http://www.rackspace.com/cloud/queues/
+.. _S3: http://aws.amazon.com/s3/
+.. _SQS: http://aws.amazon.com/sqs/
+
+"""
+
+from __future__ import absolute_import
+
+from slimta.queue import QueueError, QueueStorage
+from slimta import logging
+
+__all__ = ['CloudStorageError', 'CloudStorage']
+
+log = logging.getQueueStorageLogger(__name__)
+
+
+class CloudStorageError(QueueError):
+    """Base exception for all exceptions in the package.
+
+    """
+    pass
+
+
+class CloudStorage(QueueStorage):
+    """This class implements a :class:`~slimta.queue.QueueStorage` backend that
+    uses cloud services to store messages. It coordinates the storage of
+    messages and metadata (using `Cloud Files`_ or `S3`_) with the optional
+    message queue mechanisms (using `Cloud Queues`_ or `SQS`_) that can alert
+    other *slimta* processes that a new message is available in the object
+    store.
+
+    :param object_store: The object used as the backend for storing message
+                         contents and metadata in the cloud. Currently this can
+                         be an instance of
+                         :class:`~rackspace.RackspaceCloudFiles` or
+                         :class:`~aws.SimpleStorageService`.
+    :param message_queue: The optional object used
+                          as the backend for alerting other processes that a
+                          new message is in the object store. Currently this
+                          can be an instance of
+                          :class:`~rackspace.RackspaceCloudQueues` or
+                          :class:`~aws.SimpleQueueService`.
+
+    """
+
+    def __init__(self, object_store, message_queue=None):
+        super(CloudStorage, self).__init__()
+        self.obj_store = object_store
+        self.msg_queue = message_queue
+
+    def write(self, envelope, timestamp):
+        storage_id = self.obj_store.write_message(envelope, timestamp)
+        if self.msg_queue:
+            try:
+                self.msg_queue.queue_message(storage_id, timestamp)
+            except Exception:
+                logging.log_exception(__name__)
+        log.write(storage_id, envelope)
+        return storage_id
+
+    def set_timestamp(self, id, timestamp):
+        self.obj_store.set_message_meta(id, timestamp=timestamp)
+        log.update_meta(id, timestamp=timestamp)
+
+    def increment_attempts(self, id):
+        meta = self.obj_store.get_message_meta(id)
+        new_attempts = meta['attempts'] + 1
+        self.obj_store.set_message_meta(id, attempts=new_attempts)
+        log.update_meta(id, attempts=new_attempts)
+        return new_attempts
+
+    def set_recipients_delivered(self, id, rcpt_indexes):
+        meta = self.obj_store.get_message_meta(id)
+        current = meta.get('delivered_indexes', [])
+        new = current + rcpt_indexes
+        self.obj_store.set_message_meta(id, delivered_indexes=new)
+        log.update_meta(id, delivered_indexes=rcpt_indexes)
+
+    def load(self):
+        return self.obj_store.list_messages()
+
+    def get(self, id):
+        envelope, meta = self.obj_store.get_message(id)
+        delivered_rcpts = meta.get('delivered_indexes', [])
+        self._remove_delivered_rcpts(envelope, delivered_rcpts)
+        return envelope, meta.get('attempts', 0)
+
+    def remove(self, id):
+        self.obj_store.delete_message(id)
+        log.remove(id)
+
+    def wait(self):
+        if self.msg_queue:
+            for timestamp, storage_id, message_id in self.msg_queue.poll():
+                yield (timestamp, storage_id)
+                self.msg_queue.delete(message_id)
+            self.msg_queue.sleep()
+        else:
+            raise NotImplementedError()
+
+
+# vim:et:fdm=marker:sts=4:sw=4:ts=4