periodic_tasks

docker-compose.yml

@@ -59,9 +59,17 @@ services:
 
   beat:
     restart: always
+    environment:
+      <<: *postgres-variables
+      <<: *django-variables
     build:
       context: .
-    entrypoint: [ "celery", "-A", "postamates.celery:app", "beat" ]
+    entrypoint: [ "celery", "-A", "postamates", "beat","-l", "info", "--scheduler", "django_celery_beat.schedulers:DatabaseScheduler" ]
+    depends_on:
+      - db
+      - loyalty-rabbit
+      - django
+      - worker
   worker:
     restart: always
     environment:

postamates/celery.py

@@ -12,15 +12,15 @@ app = Celery(
     include=['service.tasks'], )
 
 app.config_from_object('django.conf:settings', namespace=CELERY_NAMESPACE)
+app.conf.beat_schedule = {
+    'age_day_every_24h': {
+        'task': 'service.tasks.add_age_day',
+        'schedule': 24 * 60 * 60.0,
+    },
+    'ml_func_every_24h': {
+        'task': 'service.tasks.raschet',
+        'schedule': 24 * 60 * 60.0,
+    },
+}
 
 app.autodiscover_tasks()
-
-# заносим таски в очередь
-# app.conf.beat_schedule = {
-#     'every': {
-#         'task': 'service.tasks.raschet',
-#         'schedule': crontab(),#периодичность (По дефолту каждую минуту)
-#         'args':[5]
-#     },
-#
-# }

requirements.txt

@@ -5,6 +5,7 @@ billiard==3.6.4.0
 catboost==1.1.1
 celery==5.2.7
 certifi==2022.12.7
+cfgv==3.3.1
 charset-normalizer==3.0.1
 click==8.1.3
 click-didyoumean==0.3.0
@@ -14,6 +15,7 @@ cligj==0.7.2
 coreapi==2.3.3
 coreschema==0.0.4
 cycler==0.11.0
+distlib==0.3.6
 Django==3.2
 django-celery-beat==2.4.0
 django-cors-headers==3.12.0
@@ -23,6 +25,7 @@ django-timezone-field==5.0
 djangorestframework==3.11.1
 drf-yasg==1.21.5
 et-xmlfile==1.1.0
+filelock==3.9.0
 Fiona==1.9.1
 fonttools==4.38.0
 future==0.18.3
@@ -31,6 +34,7 @@ geopandas==0.12.2
 graphviz==0.20.1
 greenlet==2.0.2
 gunicorn==19.9.0
+identify==2.5.20
 idna==3.4
 inflection==0.5.1
 itypes==1.2.0
@@ -40,14 +44,16 @@ kombu==5.2.4
 MarkupSafe==2.1.2
 matplotlib==3.5.2
 munch==2.5.0
+nodeenv==1.7.0
 numpy==1.24.2
 openpyxl==3.0.10
 packaging==23.0
 pandas==1.4.2
 Pillow==9.4.0
+platformdirs==3.1.1
 plotly==5.13.0
+pre-commit==3.1.1
 prompt-toolkit==3.0.37
-pre-commit
 psycopg2-binary==2.9.3
 pygeos==0.14
 pyparsing==3.0.9
@@ -56,6 +62,7 @@ pyshp==2.3.0
 python-crontab==2.7.1
 python-dateutil==2.8.2
 pytz==2022.7.1
+PyYAML==6.0
 requests==2.28.2
 ruamel.yaml==0.17.21
 ruamel.yaml.clib==0.2.7
@@ -71,7 +78,7 @@ tzdata==2022.7
 uritemplate==4.1.1
 urllib3==1.26.14
 vine==5.0.0
+virtualenv==20.20.0
 wcwidth==0.2.6
 xlrd==1.2.0
 XlsxWriter==3.0.8
-

service/service.py

@@ -7,6 +7,7 @@ from django.db.models import F
 from postamates.settings import DEFAULT_PLACEMENT_POINT_UPDATE_RADIUS
 from service import models
 from service.enums import PointStatus
+from service.tasks import raschet
 
 
 class PointService:
@@ -27,12 +28,14 @@ class PointService:
                         geometry__distance_lt=(point.geometry, Distance(m=DEFAULT_PLACEMENT_POINT_UPDATE_RADIUS)),
                     )
                     pnts.update(target_post_cnt=F('target_post_cnt') + 1)
+                    raschet.delay()
             elif new_status == PointStatus.Cancelled.name or new_status == PointStatus.Pending.name:
                 if point.status == PointStatus.Installation.name:
                     pnts = models.PlacementPoint.objects.filter(
                         geometry__distance_lt=(point.geometry, Distance(m=DEFAULT_PLACEMENT_POINT_UPDATE_RADIUS)),
                     )
                     pnts.update(target_post_cnt=F('target_post_cnt') - 1 if F('target_post_cnt') != 0 else 0)
+                    raschet.delay()
 
     @staticmethod
     def update_status(qs: models.PlacementPoint, new_status: str) -> models.PlacementPoint:

service/tasks.py

@@ -1,6 +1,9 @@
 import time
 
 from celery import shared_task
+from django.db.models import F
+
+from service.models import PlacementPoint
 
 
 # Запустить worker
@@ -9,7 +12,19 @@ from celery import shared_task
 # celery -A postamates beat -l INFO.
 
 @shared_task()
-def raschet(sec: int):
+def raschet():
     print('Hello. Celery task is running...')
-    time.sleep(sec)
+    time.sleep(5)
     print('Finish')
+
+
+@shared_task()
+def add_age_day():
+    qs = PlacementPoint.objects
+    c1 = qs.filter(sample_trn=True).count()
+    qs.update(age_day=F('age_day') + 1)
+    qs2 = qs.filter(age_day__gt=270)
+    qs2.update(sample_trn=True)
+    c2 = PlacementPoint.objects.filter(sample_trn=True).count()
+    if c2 - c1 != 0:
+        raschet.delay()

service/views.py

@@ -190,7 +190,7 @@ class PlacementPointViewSet(ReadOnlyModelViewSet):
 
     @action(detail=False, methods=['get'])
     def start(self, request):
-        raschet.delay(5)
+        raschet.delay()
         return Response('Sucess', status=HTTPStatus.OK)