move datasync tasks into services/datasync.py

src/pretix/base/services/datasync.py

@@ -0,0 +1,83 @@
+#
+# This file is part of pretix (Community Edition).
+#
+# Copyright (C) 2014-2020 Raphael Michel and contributors
+# Copyright (C) 2020-2021 rami.io GmbH and contributors
+#
+# This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General
+# Public License as published by the Free Software Foundation in version 3 of the License.
+#
+# ADDITIONAL TERMS APPLY: Pursuant to Section 7 of the GNU Affero General Public License, additional terms are
+# applicable granting you additional permissions and placing additional restrictions on your usage of this software.
+# Please refer to the pretix LICENSE file to obtain the full terms applicable to this work. If you did not receive
+# this file, see <https://pretix.eu/about/en/license>.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied
+# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Affero General Public License along with this program.  If not, see
+# <https://www.gnu.org/licenses/>.
+#
+
+import logging
+from datetime import timedelta
+from itertools import groupby
+
+from django.db.models import F, Window
+from django.db.models.functions import RowNumber
+from django.dispatch import receiver
+from django.utils.timezone import now
+from django_scopes import scope, scopes_disabled
+
+from pretix.base.datasync.datasync import sync_targets
+from pretix.base.models.datasync import OrderSyncQueue
+from pretix.base.signals import periodic_task
+from pretix.celery_app import app
+
+logger = logging.getLogger(__name__)
+
+
+@receiver(periodic_task, dispatch_uid="data_sync_periodic_sync_all")
+def periodic_sync_all(sender, **kwargs):
+    sync_all.apply_async()
+
+
+@receiver(periodic_task, dispatch_uid="data_sync_periodic_reset_in_flight")
+def periodic_reset_in_flight(sender, **kwargs):
+    for sq in OrderSyncQueue.objects.filter(
+        in_flight=True,
+        in_flight_since__lt=now() - timedelta(minutes=20),
+    ):
+        sq.set_sync_error('timeout', [], 'Timeout')
+
+
+@app.task()
+def sync_all():
+    with scopes_disabled():
+        queue = (
+            OrderSyncQueue.objects
+            .filter(
+                in_flight=False,
+                not_before__lt=now(),
+                need_manual_retry__isnull=True,
+            )
+            .order_by(Window(
+                expression=RowNumber(),
+                partition_by=[F("event_id")],
+                order_by="not_before",
+            ))
+            .prefetch_related("event")
+            [:1000]
+        )
+        grouped = groupby(sorted(queue, key=lambda q: (q.sync_provider, q.event.pk)), lambda q: (q.sync_provider, q.event))
+        for (target, event), queued_orders in grouped:
+            target_cls, meta = sync_targets.get(identifier=target, active_in=event)
+
+            if not target_cls:
+                # sync plugin not found (plugin deactivated or uninstalled) -> drop outstanding jobs
+                OrderSyncQueue.objects.filter(pk__in=[sq.pk for sq in queued_orders]).delete()
+
+            with scope(organizer=event.organizer):
+                with target_cls(event=event) as p:
+                    p.sync_queued_orders(queued_orders)