.gitlab-ci.yml

@@ -82,9 +82,9 @@ python linting:
     - source env/bin/activate
     - pip install -r requirements/gitlab.txt
   script:
-    - isort -rc src/ --check-only
-    - black -l 88 --check src/
-    - autoflake --check --remove-all-unused-imports --ignore-init-module-imports --recursive src/
+    - isort src/ --check-only --recursive
+    - black src/ --line-length 88 --check
+    - autoflake src/ --check --recursive --remove-all-unused-imports --ignore-init-module-imports
 
 deploy:
   stage: deploy

requirements/dev.txt

@@ -6,5 +6,5 @@ django-debug-toolbar==2.0
 django-extensions==2.1.9
 
 black==19.3b0
-isort==4.3.20
-autoflake==1.3
+isort==4.3.21
+autoflake==1.3.1

src/newsreader/news/collection/tasks.py

@@ -3,16 +3,19 @@ from django.core.exceptions import ObjectDoesNotExist
 from newsreader.accounts.models import User
 from newsreader.celery import app
 from newsreader.news.collection.feed import FeedCollector
+from newsreader.utils.celery import MemCacheLock
 
 
-@app.task
-def collect(user_pk):
+@app.task(bind=True)
+def collect(self, user_pk):
     try:
         user = User.objects.get(pk=user_pk)
     except ObjectDoesNotExist:
         return
 
-    rules = user.rules.all()
+    with MemCacheLock(f"{user.email}-task", self.app.oid) as acquired:
+        if acquired:
+            rules = user.rules.all()
 
-    collector = FeedCollector()
-    collector.collect(rules=rules)
+            collector = FeedCollector()
+            collector.collect(rules=rules)

src/newsreader/utils/celery.py

@@ -0,0 +1,22 @@
+from django.core.cache import cache
+
+from celery.five import monotonic
+
+
+LOCK_EXPIRE = 60 * 10  # 10 minutes
+
+
+class MemCacheLock:
+    def __init__(self, lock_id, oid):
+        self.lock_id = lock_id
+        self.oid = oid
+
+        self.timeout_at = monotonic() + LOCK_EXPIRE - 3
+
+    def __enter__(self):
+        self.status = cache.add(self.lock_id, self.oid, LOCK_EXPIRE)
+        return self.status
+
+    def __exit__(self, *args, **kwargs):
+        if monotonic() < self.timeout_at and self.status:
+            cache.delete(self.lock_id)