Pull upstream PyQS changes into our repo

.github/workflows/ci.yml

@@ -0,0 +1,26 @@
+name: CI
+
+on: [push]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    env:
+      VIRTUAL_ENV: ignore
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.9
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r development.txt
+      - name: Lint
+        run: |
+          pip install flake8
+          python -m flake8 pyqs tests
+      - name: Test
+        run: |
+          make test

.gitignore

@@ -6,6 +6,7 @@ dist/*
 *.egg-info/*
 build/*
 htmlcov/*
+.idea/*
 *~
 *#
 

.travis.yml

@@ -1,7 +1,6 @@
 language: python
 
 python:
-  - "2.7"
   - "3.6"
   - "3.7"
   - "3.8"

CHANGELOG.rst

@@ -1,6 +1,25 @@
 Changelog
 ---------
 
+1.0.1
+~~~~~
+- Add MessageId for tracking task executions
+
+1.0.0
+~~~~~
+- Drop Py2 support
+- Add new SimpleProcessWorker (https://github.com/spulec/PyQS/pull/76)
+
+0.1.6
+~~~~~
+
+- Fix broken pickle of botocore clients.
+
+0.1.5
+~~~~~
+
+- Add events hooks for pre and post processors.
+
 0.1.4
 ~~~~~
 

Makefile

@@ -41,26 +41,7 @@ clean:
 	@find . -name __pycache__ -delete
 	@rm -rf .coverage *.egg-info *.log build dist MANIFEST yc
 
-publish: clean tag
-	@if [ -e "$$HOME/.pypirc" ]; then \
-		echo "Uploading to '$(CUSTOM_PIP_INDEX)'"; \
-		python setup.py register -r "$(CUSTOM_PIP_INDEX)"; \
-		python setup.py sdist upload -r "$(CUSTOM_PIP_INDEX)"; \
-	else \
-		echo "You should create a file called '.pypirc' under your home dir.\n"; \
-		echo "That's the right place to configure 'pypi' repos.\n"; \
-		exit 1; \
-	fi
-
-tag:
-	@if [ $$(git rev-list $$(git describe --abbrev=0 --tags)..HEAD --count) -gt 0 ]; then \
-		if [ $$(git log  -n 1 --oneline $$(git describe --abbrev=0 --tags)..HEAD CHANGELOG.rst | wc -l) -gt 0 ]; then \
-			git tag $$(python setup.py --version) && git push --tags || echo 'Version already released, update your version!'; \
-		else \
-			echo "CHANGELOG not updated since last release!"; \
-			exit 1; \
-		fi; \
-	else \
-		echo "No commits since last release!"; \
-		exit 1;\
-	fi
+publish: clean
+	rm -rf dist
+	python -m pep517.build --source --binary .
+	twine upload dist/*

README.rst

@@ -78,7 +78,7 @@ To read tasks we need to run PyQS. If the task is already in your
 
     $ pyqs email.tasks.send_email
 
-If we want want to run all tasks with a certain prefix. This is based on
+If we want to run all tasks with a certain prefix. This is based on
 Python's `fnmatch <http://docs.python.org/2/library/fnmatch.html>`__.
 
 .. code:: bash
@@ -100,6 +100,45 @@ messages.
 
     $ pyqs send_email --concurrency 10
 
+Simple Process Worker
+~~~~~~~~~~~~~~~~~~~~~
+
+To use a simpler version of PyQS that deals with some of the edge cases in the original implementation, pass the ``simple-worker`` flag.
+
+.. code:: bash
+
+    $ pyqs send_email --simple-worker
+
+The Simple Process Worker differs in the following way from the original implementation.
+
+* Does not use an internal queue and removes support for the ``prefetch-multiplier`` flag. This helps simply the mental model required, as messages are not on both the SQS queue and an internal queue.
+* When the ``simple-worker`` flag is passed, the default ``batchsize`` is 1 instead of 10. This is configurable.
+* Does not check the visibility timeout when reading or processing a message from SQS.
+    * Allowing the worker to process the message even past its visibility timeout means we solve the problem of never processing a message if ``max_receives=1`` and we incorrectly set a shorter visibility timeout and exceed the visibility timeout. Previously, this message would have ended up in the DLQ, if one was configured, and never actually processed.
+    * It increases the probability that we process a message more than once, especially if ``batchsize > 1``, but this can be solved by the developer checking if the message has already been processed.
+
+Hooks
+~~~~~
+
+PyQS has an event registry which can be used to run a function before or after every tasks runs.
+
+.. code:: python
+
+    from pyqs import task, events
+
+    def print_pre_process(context):
+        print({"pre_process": context})
+
+    def print_post_process(context):
+        print({"post_process": context})
+
+    events.register_event("pre_process", print_pre_process)
+    events.register_event("post_process", print_post_process)
+
+    @task(queue="my_queue")
+    def send_email(subject):
+        pass
+
 Operational Notes
 ~~~~~~~~~~~~~~~~~
 

development.txt

@@ -2,7 +2,9 @@ coverage==4.4.1
 mock==1.0.1
 moto==1.3.13
 nose==1.3.0
+pep517==0.9.1
 pre-commit==0.7.6
 sure==1.2.2
+twine==3.3.0
 functools32;python_version=='2.7'
 pycodestyle==2.4.0

pyproject.toml

@@ -0,0 +1,3 @@
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"

pyqs/__init__.py

@@ -1,4 +1,4 @@
 from .decorator import task  # noqa
 
 __title__ = 'pyqs'
-__version__ = '0.1.4'
+__version__ = '1.0.1'

pyqs/events.py

@@ -0,0 +1,43 @@
+"""
+pyqs events registry: register callback functions on pyqs events
+
+Usage:
+from pyqs.events import register_event
+
+register_event("pre_process", lambda context: print(context))
+"""
+
+
+class Events:
+    def __init__(self):
+        self.pre_process = []
+        self.post_process = []
+
+    def clear(self):
+        self.pre_process = []
+        self.post_process = []
+
+
+# Global singleton
+_EVENTS = Events()
+
+
+class NoEventException(Exception):
+    pass
+
+
+def register_event(name, callback):
+    if hasattr(_EVENTS, name):
+        getattr(_EVENTS, name).append(callback)
+    else:
+        raise NoEventException(
+            "{name} is not a valid pyqs event.".format(name=name)
+        )
+
+
+def get_events():
+    return _EVENTS
+
+
+def clear_events():
+    _EVENTS.clear()

pyqs/main.py

@@ -7,11 +7,28 @@
 import sys
 from argparse import ArgumentParser
 
-from .worker import ManagerWorker
+from .worker import ManagerWorker, SimpleManagerWorker
 from . import __version__
 
 logger = logging.getLogger("pyqs")
 
+SIMPLE_WORKER_DEFAULT_BATCH_SIZE = 1
+DEFAULT_BATCH_SIZE = 10
+
+
+def _set_batchsize(args):
+    batchsize = args.batchsize
+    if batchsize:
+        return batchsize
+
+    simple_worker = args.simple_worker
+    if simple_worker:
+        # Default batchsize for SimpleProcessWorker
+        return SIMPLE_WORKER_DEFAULT_BATCH_SIZE
+
+    # Default batchsize for ProcessWorker
+    return DEFAULT_BATCH_SIZE
+
 
 def main():
     parser = ArgumentParser(description="""
@@ -77,6 +94,15 @@ def main():
         action="store",
     )
 
+    parser.add_argument(
+        "--endpoint-url",
+        dest="endpoint_url",
+        type=str,
+        default=None,
+        help="AWS SQS endpoint url",
+        action="store",
+    )
+
     parser.add_argument(
         "--interval",
         dest="interval",
@@ -90,7 +116,7 @@ def main():
         "--batchsize",
         dest="batchsize",
         type=int,
-        default=10,
+        default=None,
         help='How many messages to download at a time from SQS.',
         action="store",
     )
@@ -107,6 +133,13 @@ def main():
         action="store",
     )
 
+    parser.add_argument(
+        '--simple-worker',
+        dest='simple_worker',
+        default=False,
+        action='store_true'
+    )
+
     args = parser.parse_args()
 
     _main(
@@ -117,8 +150,10 @@ def main():
         access_key_id=args.access_key_id,
         secret_access_key=args.secret_access_key,
         interval=args.interval,
-        batchsize=args.batchsize,
-        prefetch_multiplier=args.prefetch_multiplier
+        batchsize=_set_batchsize(args),
+        prefetch_multiplier=args.prefetch_multiplier,
+        simple_worker=args.simple_worker,
+        endpoint_url=args.endpoint_url,
     )
 
 
@@ -130,17 +165,29 @@ def _add_cwd_to_path():
 
 def _main(queue_prefixes, concurrency=5, logging_level="WARN",
           region=None, access_key_id=None, secret_access_key=None,
-          interval=1, batchsize=10, prefetch_multiplier=2):
+          interval=1, batchsize=DEFAULT_BATCH_SIZE, prefetch_multiplier=2,
+          simple_worker=False, endpoint_url=None):
     logging.basicConfig(
         format="[%(levelname)s]: %(message)s",
         level=getattr(logging, logging_level),
     )
     logger.info("Starting PyQS version {}".format(__version__))
-    manager = ManagerWorker(
-        queue_prefixes, concurrency, interval, batchsize,
-        prefetch_multiplier=prefetch_multiplier, region=region,
-        access_key_id=access_key_id, secret_access_key=secret_access_key,
-    )
+
+    if simple_worker:
+        manager = SimpleManagerWorker(
+            queue_prefixes, concurrency, interval, batchsize,
+            region=region, access_key_id=access_key_id,
+            secret_access_key=secret_access_key,
+            endpoint_url=endpoint_url,
+        )
+    else:
+        manager = ManagerWorker(
+            queue_prefixes, concurrency, interval, batchsize,
+            prefetch_multiplier=prefetch_multiplier, region=region,
+            access_key_id=access_key_id, secret_access_key=secret_access_key,
+            endpoint_url=endpoint_url,
+        )
+
     _add_cwd_to_path()
     manager.start()
     manager.sleep()