Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add continuous commerce synchronization

Browse Source
This commit is contained in:
lizerk 2026-08-17 19:31:56 +03:00
parent 9610980014
commit b622370067
10 changed files with 257 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
backend/api/v1/commerce/serializers.py
View File

@ -1,3 +1,6 @@
from datetime import timedelta
from django.utils import timezone
from rest_framework import serializers
from commerce.models import (
@ -81,6 +84,9 @@ class IntegrationConnectionSerializer(serializers.ModelSerializer):
"last_sync_finished_at",
"last_stats",
"task_id",
"auto_sync",
"sync_interval_minutes",
"next_sync_at",
"recent_runs",
"created_at",
"updated_at",
@ -94,6 +100,7 @@ class IntegrationConnectionSerializer(serializers.ModelSerializer):
"last_sync_finished_at",
"last_stats",
"task_id",
"next_sync_at",
"created_at",
"updated_at",
]
@ -163,13 +170,22 @@ class IntegrationConnectionSerializer(serializers.ModelSerializer):
)
def update(self, instance, validated_data):
previous_auth_type = instance.auth_type
credentials = self._credentials(validated_data)
if credentials:
instance.encrypted_credentials = encrypt_credentials(credentials)
for field, value in validated_data.items():
setattr(instance, field, value)
instance.status = "disconnected"
instance.last_error = ""
credentials_changed = credentials or instance.auth_type != previous_auth_type
if credentials_changed:
instance.status = "disconnected"
instance.last_error = ""
if "auto_sync" in validated_data or "sync_interval_minutes" in validated_data:
instance.next_sync_at = (
timezone.now() + timedelta(minutes=instance.sync_interval_minutes)
if instance.auto_sync
else None
)
instance.save()
return instance

26
backend/api/v1/commerce/viewsets.py
View File

@ -1,4 +1,6 @@
from django.utils import timezone
from datetime import timedelta
from django.db.models import Count
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import filters, viewsets
@ -25,7 +27,7 @@ from commerce.services import (
)
from commerce.services.credentials import CredentialError
from commerce.integrations.moysklad import MoySkladError
from commerce.tasks import sync_moysklad_run
from commerce.tasks import enqueue_connection_sync
from .permissions import CommerceAccessPermission
from .serializers import (
@ -71,6 +73,11 @@ class IntegrationConnectionViewSet(viewsets.ModelViewSet):
connection.account_name = context["account_name"]
connection.last_error = ""
connection.last_tested_at = timezone.now()
connection.next_sync_at = (
timezone.now() + timedelta(minutes=connection.sync_interval_minutes)
if connection.auto_sync
else None
)
connection.source.metadata = {
**connection.source.metadata,
"account_id": context["account_id"],
@ -82,6 +89,7 @@ class IntegrationConnectionViewSet(viewsets.ModelViewSet):
"account_name",
"last_error",
"last_tested_at",
"next_sync_at",
"updated_at",
]
)
@ -95,16 +103,12 @@ class IntegrationConnectionViewSet(viewsets.ModelViewSet):
{"detail": "Синхронизация уже выполняется."},
status=status.HTTP_409_CONFLICT,
)
run = IntegrationSyncRun.objects.create(connection=connection)
connection.status = IntegrationStatus.QUEUED
connection.task_id = ""
connection.last_error = ""
connection.save(update_fields=["status", "task_id", "last_error", "updated_at"])
async_result = sync_moysklad_run.delay(str(run.pk))
run.task_id = async_result.id or ""
run.save(update_fields=["task_id"])
connection.task_id = run.task_id
connection.save(update_fields=["task_id", "updated_at"])
run = enqueue_connection_sync(connection)
if run is None:
return Response(
{"detail": "Синхронизация уже выполняется."},
status=status.HTTP_409_CONFLICT,
)
return Response(
IntegrationSyncRunSerializer(run).data,
status=status.HTTP_202_ACCEPTED,

4
backend/commerce/README.md
View File

@ -33,6 +33,10 @@ created by employees. The first production connector is `moysklad`: it scans
the complete paginated assortment, aggregate stock, shipments and customer
returns. Raw API payloads are retained in `SourceRecord` before normalization.
Repeated scans update the same source identities and append stock observations.
Celery Beat checks due connections every minute. Each connection stores whether
automatic synchronization is enabled, its interval (15 minutes by default) and
the next planned run; row locking and connection status prevent overlapping
full scans. Both a Celery worker and Celery Beat must be running.
Imported cards are matched by an existing source binding, SKU or barcode. A
card that cannot be matched safely is created separately and can later be

29
backend/commerce/migrations/0003_integrationconnection_auto_sync_and_more.py Normal file
View File

@ -0,0 +1,29 @@
# Generated by Django 5.2.4 on 2026-08-17 16:26
import django.core.validators
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('commerce', '0002_integrationconnection_integrationsyncrun'),
]
operations = [
migrations.AddField(
model_name='integrationconnection',
name='auto_sync',
field=models.BooleanField(default=True, verbose_name='Автоматическая синхронизация'),
),
migrations.AddField(
model_name='integrationconnection',
name='next_sync_at',
field=models.DateTimeField(blank=True, db_index=True, null=True, verbose_name='Следующая синхронизация'),
),
migrations.AddField(
model_name='integrationconnection',
name='sync_interval_minutes',
field=models.PositiveIntegerField(default=15, validators=[django.core.validators.MinValueValidator(5), django.core.validators.MaxValueValidator(1440)], verbose_name='Интервал синхронизации, минут'),
),
]

10
backend/commerce/models/source.py
View File

@ -1,6 +1,7 @@
import uuid
from django.conf import settings
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
@ -159,6 +160,15 @@ class IntegrationConnection(models.Model):
)
last_stats = models.JSONField("Результат синхронизации", default=dict, blank=True)
task_id = models.CharField("ID фоновой задачи", max_length=255, blank=True)
auto_sync = models.BooleanField("Автоматическая синхронизация", default=True)
sync_interval_minutes = models.PositiveIntegerField(
"Интервал синхронизации, минут",
default=15,
validators=[MinValueValidator(5), MaxValueValidator(1440)],
)
next_sync_at = models.DateTimeField(
"Следующая синхронизация", null=True, blank=True, db_index=True
)
created_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,

87
backend/commerce/tasks/__init__.py
View File

@ -1,17 +1,69 @@
import logging
from datetime import timedelta
from celery import shared_task
from django.db import transaction
from django.db.models import Q
from django.utils import timezone
from commerce.integrations import sync_registered_source
from commerce.integrations.moysklad import MoySkladError
from commerce.models import IntegrationStatus, IntegrationSyncRun
from commerce.models import IntegrationConnection, IntegrationStatus, IntegrationSyncRun
from commerce.services import sync_moysklad_connection
from commerce.services.credentials import CredentialError
logger = logging.getLogger(__name__)
def _next_sync_at(connection, *, now=None):
if not connection.auto_sync:
return None
return (now or timezone.now()) + timedelta(minutes=connection.sync_interval_minutes)
def enqueue_connection_sync(connection):
"""Create and publish one run while preventing overlapping scans."""
with transaction.atomic():
locked = IntegrationConnection.objects.select_for_update().get(pk=connection.pk)
if locked.status in {IntegrationStatus.QUEUED, IntegrationStatus.SYNCING}:
return None
run = IntegrationSyncRun.objects.create(connection=locked)
locked.status = IntegrationStatus.QUEUED
locked.task_id = ""
locked.last_error = ""
locked.next_sync_at = _next_sync_at(locked)
locked.save(
update_fields=[
"status",
"task_id",
"last_error",
"next_sync_at",
"updated_at",
]
)
try:
async_result = sync_moysklad_run.delay(str(run.pk))
except Exception:
retry_at = timezone.now() + timedelta(minutes=5)
IntegrationSyncRun.objects.filter(pk=run.pk).update(
status="error",
error="Не удалось поставить синхронизацию в очередь.",
finished_at=timezone.now(),
)
IntegrationConnection.objects.filter(pk=connection.pk).update(
status=IntegrationStatus.ERROR,
last_error="Не удалось поставить синхронизацию в очередь.",
next_sync_at=retry_at,
)
raise
task_id = async_result.id or ""
IntegrationSyncRun.objects.filter(pk=run.pk).update(task_id=task_id)
IntegrationConnection.objects.filter(pk=connection.pk).update(task_id=task_id)
run.task_id = task_id
return run
@shared_task(name="commerce.sync_source")
def sync_source(source_code: str, entity_type: str):
results = sync_registered_source(source_code, entity_type=entity_type)
@ -54,11 +106,13 @@ def sync_moysklad_run(self, run_id: str):
connection.status = IntegrationStatus.ERROR
connection.last_error = message
connection.last_sync_finished_at = finished_at
connection.next_sync_at = _next_sync_at(connection, now=finished_at)
connection.save(
update_fields=[
"status",
"last_error",
"last_sync_finished_at",
"next_sync_at",
"updated_at",
]
)
@ -85,13 +139,44 @@ def sync_moysklad_run(self, run_id: str):
connection.last_stats = stats
connection.last_error = ""
connection.last_sync_finished_at = finished_at
connection.next_sync_at = _next_sync_at(connection, now=finished_at)
connection.save(
update_fields=[
"status",
"last_stats",
"last_error",
"last_sync_finished_at",
"next_sync_at",
"updated_at",
]
)
return stats
@shared_task(name="commerce.dispatch_due_connection_syncs")
def dispatch_due_connection_syncs():
now = timezone.now()
stale_cutoff = now - timedelta(minutes=45)
IntegrationConnection.objects.filter(
auto_sync=True,
status__in=[IntegrationStatus.QUEUED, IntegrationStatus.SYNCING],
updated_at__lte=stale_cutoff,
).update(
status=IntegrationStatus.ERROR,
last_error="Предыдущая синхронизация была прервана и будет перезапущена.",
next_sync_at=now,
updated_at=now,
)
due_ids = list(
IntegrationConnection.objects.filter(
auto_sync=True,
status__in=[IntegrationStatus.READY, IntegrationStatus.ERROR],
)
.filter(Q(next_sync_at__isnull=True) | Q(next_sync_at__lte=now))
.values_list("pk", flat=True)
)
queued = 0
for connection in IntegrationConnection.objects.filter(pk__in=due_ids):
if enqueue_connection_sync(connection) is not None:
queued += 1
return {"queued": queued}

4
backend/eusrr_backend/settings.py
View File

@ -762,6 +762,10 @@ CELERYD_PIDFILE = os.path.join(tempfile.gettempdir(), "celery-worker.pid")
# Расписание периодических задач
CELERY_BEAT_SCHEDULE = {
"commerce-sync-due-connections": {
"task": "commerce.dispatch_due_connection_syncs",
"schedule": 60.0,
},
"open-monthly-payroll-period": {
"task": "finance.tasks.open_monthly_payroll_period",
"schedule": crontab(day_of_month="1", hour=0, minute=5),

53
backend/tests/commerce/test_connections.py
View File

@ -1,3 +1,4 @@
from datetime import timedelta
from unittest.mock import Mock, patch
import httpx
@ -21,6 +22,7 @@ from commerce.services.moysklad_sync import (
normalize_moysklad_assortment_row,
normalize_moysklad_document,
)
from commerce.tasks import dispatch_due_connection_syncs
pytestmark = pytest.mark.django_db
@ -100,9 +102,7 @@ def test_connection_sync_action_queues_background_run(api_client, user_factory):
)
async_result = Mock(id="celery-task-1")
with patch(
"api.v1.commerce.viewsets.sync_moysklad_run.delay", return_value=async_result
):
with patch("commerce.tasks.sync_moysklad_run.delay", return_value=async_result):
response = api_client.post(
reverse(
"api:v1:commerce:connection-sync",
@ -117,6 +117,53 @@ def test_connection_sync_action_queues_background_run(api_client, user_factory):
assert connection.task_id == "celery-task-1"
def test_beat_dispatcher_queues_due_connection():
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
connection = IntegrationConnection.objects.create(
source=source,
provider="moysklad",
encrypted_credentials="unused",
status=IntegrationStatus.READY,
auto_sync=True,
sync_interval_minutes=15,
next_sync_at=timezone.now() - timedelta(minutes=1),
)
async_result = Mock(id="scheduled-task-1")
with patch("commerce.tasks.sync_moysklad_run.delay", return_value=async_result):
result = dispatch_due_connection_syncs()
connection.refresh_from_db()
assert result == {"queued": 1}
assert connection.status == IntegrationStatus.QUEUED
assert connection.task_id == "scheduled-task-1"
assert connection.next_sync_at > timezone.now()
def test_schedule_update_does_not_disconnect_connection(api_client, user_factory):
user = user_factory(staff=True)
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
connection = IntegrationConnection.objects.create(
source=source,
provider="moysklad",
encrypted_credentials="encrypted",
status=IntegrationStatus.READY,
)
api_client.force_authenticate(user=user)
response = api_client.patch(
reverse("api:v1:commerce:connection-detail", args=[connection.pk]),
{"auto_sync": True, "sync_interval_minutes": 60},
format="json",
)
assert response.status_code == status.HTTP_200_OK
connection.refresh_from_db()
assert connection.status == IntegrationStatus.READY
assert connection.sync_interval_minutes == 60
assert connection.next_sync_at is not None
def test_moysklad_row_creates_product_binding_raw_record_and_stock():
source = SourceSystem.objects.create(code="moysklad", name="МойСклад")
connection = IntegrationConnection.objects.create(

36
frontend/src/app/products/connections/page.tsx
View File

@ -11,6 +11,7 @@ import {
ArrowLeft,
CheckCircle2,
CloudDownload,
Clock3,
KeyRound,
Loader2,
PackageCheck,
@ -146,6 +147,34 @@ export default function ProductConnectionsPage() {
}
}
async function updateSchedule(
connection: IntegrationConnection,
changes: Pick<
IntegrationConnectionInput,
"auto_sync" | "sync_interval_minutes"
>,
) {
setBusy(connection.id);
setError(null);
try {
const updated = await apiClient.updateCommerceConnection(
connection.id,
changes,
);
setConnections((current) =>
current.map((item) => (item.id === updated.id ? updated : item)),
);
} catch (requestError) {
setError(
requestError instanceof Error
? requestError.message
: "Не удалось изменить расписание.",
);
} finally {
setBusy(null);
}
}
async function disconnect(connection: IntegrationConnection) {
if (!window.confirm("Отключить сервис? Уже импортированные товары сохранятся.")) {
return;
@ -216,6 +245,13 @@ export default function ProductConnectionsPage() {
<div><dt className="text-xs app-text-muted">Последняя проверка</dt><dd className="mt-1">{displayDate(moysklad.last_tested_at)}</dd></div>
<div><dt className="text-xs app-text-muted">Последний импорт</dt><dd className="mt-1">{displayDate(moysklad.last_sync_finished_at)}</dd></div>
</dl>
<div className="flex flex-wrap items-center justify-between gap-4 rounded-xl bg-[var(--surface-secondary)] p-4">
<label className="flex items-center gap-3 text-sm"><input type="checkbox" checked={moysklad.auto_sync} disabled={Boolean(busy)} onChange={(event) => void updateSchedule(moysklad, { auto_sync: event.target.checked, sync_interval_minutes: moysklad.sync_interval_minutes })} /><span><span className="block font-medium">Постоянная синхронизация</span><span className="block text-xs app-text-muted">Сервер запускает импорт автоматически</span></span></label>
<div className="flex flex-wrap items-center gap-3">
<label className="flex items-center gap-2 text-sm app-text-muted"><Clock3 className="h-4 w-4" /><span>Каждые</span><select value={moysklad.sync_interval_minutes} disabled={!moysklad.auto_sync || Boolean(busy)} onChange={(event) => void updateSchedule(moysklad, { auto_sync: true, sync_interval_minutes: Number(event.target.value) })} className="rounded-lg border border-[var(--border-subtle)] bg-[var(--surface-primary)] px-2 py-1.5 text-sm text-[var(--foreground)] disabled:opacity-50"><option value={5}>5 минут</option><option value={15}>15 минут</option><option value={30}>30 минут</option><option value={60}>1 час</option><option value={360}>6 часов</option><option value={1440}>сутки</option></select></label>
<span className="text-xs app-text-muted">Следующий запуск: {moysklad.auto_sync ? displayDate(moysklad.next_sync_at) : "отключён"}</span>
</div>
</div>
<div className="flex flex-wrap gap-2">
<button type="button" disabled={Boolean(syncActive || busy)} onClick={() => void startSync(moysklad)} className="app-action-primary inline-flex items-center gap-2 rounded-lg px-3 py-2 text-sm disabled:opacity-50">{syncActive ? <Loader2 className="h-4 w-4 animate-spin" /> : <CloudDownload className="h-4 w-4" />} {syncActive ? "Импорт выполняется" : "Синхронизировать всё"}</button>
<button type="button" disabled={Boolean(busy)} onClick={() => { setAuthType(moysklad.auth_type); setEditing(true); }} className="app-action-secondary inline-flex items-center gap-2 rounded-lg px-3 py-2 text-sm"><KeyRound className="h-4 w-4" /> Обновить доступ</button>

5
frontend/src/types/commerce.ts
View File

@ -64,6 +64,9 @@ export type IntegrationConnection = {
sale_lines: number;
}>;
task_id: string;
auto_sync: boolean;
sync_interval_minutes: number;
next_sync_at: string | null;
recent_runs: IntegrationSyncRun[];
created_at: string;
updated_at: string;
@ -75,6 +78,8 @@ export type IntegrationConnectionInput = {
token?: string;
login?: string;
password?: string;
auto_sync?: boolean;
sync_interval_minutes?: number;
};
export type CommerceProduct = {