Sync station storage sources from the stations list, per station or all at once

ami/main/api/serializers.py

@@ -184,6 +184,7 @@ class DeploymentListSerializer(DefaultSerializer):
     device = DeviceNestedSerializer(read_only=True)
     research_site = SiteNestedSerializer(read_only=True)
     jobs = JobStatusSerializer(many=True, read_only=True)
+    data_source_connected = serializers.SerializerMethodField()
 
     class Meta:
         model = Deployment
@@ -208,8 +209,19 @@ class Meta:
             "device",
             "research_site",
             "jobs",
+            "data_source_connected",
         ]
 
+    def get_data_source_connected(self, obj: Deployment) -> bool:
+        """
+        Whether the station has a storage source configured.
+
+        The stations list uses this to show the per-row Sync button only where a
+        sync can succeed, and to count how many stations "Sync all" would cover.
+        Reads the foreign key id already on the row, so it adds no query.
+        """
+        return obj.data_source_id is not None
+
     def get_events(self, obj):
         """
         Return URL to the events endpoint filtered by this deployment.

ami/main/api/views.py

@@ -336,30 +336,69 @@ def get_queryset(self) -> QuerySet:
 
         return qs
 
+    @staticmethod
+    def _enqueue_sync_job(deployment: Deployment):
+        """
+        Create and enqueue a ``DataStorageSyncJob`` for a single deployment.
+
+        Shared by the per-row ``sync`` action and the bulk ``sync_all`` action so
+        both build the job the same way (one job per deployment, matching the
+        admin bulk action).
+        """
+        from ami.jobs.models import DataStorageSyncJob, Job
+
+        job = Job.objects.create(
+            name=f"Sync captures for deployment {deployment.pk}",
+            deployment=deployment,
+            project=deployment.project,
+            job_type_key=DataStorageSyncJob.key,
+        )
+        job.enqueue()
+        return job
+
     @action(detail=True, methods=["post"], name="sync")
     def sync(self, _request, pk=None) -> Response:
         """
         Queue a task to sync data from the deployment's data source.
         """
         deployment: Deployment = self.get_object()
         if deployment and deployment.data_source:
-            # queued_task = tasks.sync_source_images.delay(deployment.pk)
-            from ami.jobs.models import DataStorageSyncJob, Job
-
-            job = Job.objects.create(
-                name=f"Sync captures for deployment {deployment.pk}",
-                deployment=deployment,
-                project=deployment.project,
-                job_type_key=DataStorageSyncJob.key,
+            job = self._enqueue_sync_job(deployment)
+            logger.info(
+                f"Syncing captures for deployment {deployment.pk} from {deployment.data_source_uri} in background."
             )
-            job.enqueue()
-            msg = f"Syncing captures for deployment {deployment.pk} from {deployment.data_source_uri} in background."
-            logger.info(msg)
-            assert deployment.project
-            return Response({"job_id": job.pk, "project_id": deployment.project.pk})
+            return Response({"job_id": job.pk, "project_id": deployment.project_id})
         else:
             raise api_exceptions.ValidationError(detail="Deployment must have a data source to sync captures from")
 
+    @action(detail=False, methods=["post"], name="sync-all", url_path="sync-all")
+    def sync_all(self, request) -> Response:
+        """
+        Queue a sync job for every station in the project that has a storage source.
+
+        Enqueues one ``DataStorageSyncJob`` per connected station (separate jobs,
+        not one consolidated job), matching the per-row ``sync`` action and the
+        admin bulk action. Requires the ``project_id`` query parameter and the
+        sync permission on the project.
+        """
+        project = self.get_active_project()
+        if not project:
+            raise api_exceptions.ValidationError(detail="A project_id is required to sync all stations.")
+
+        # ObjectPermission.has_permission() is a no-op and has_object_permission()
+        # only runs for detail actions (via get_object()), so a detail=False action
+        # must check permissions itself. Probe the same sync permission the per-row
+        # action enforces, resolved against the project.
+        if not Deployment(project=project).check_permission(request.user, "sync"):
+            raise api_exceptions.PermissionDenied(
+                detail="You do not have permission to sync stations in this project."
+            )
+
+        deployments = self.get_queryset().filter(data_source__isnull=False)
+        job_ids = [self._enqueue_sync_job(deployment).pk for deployment in deployments]
+        logger.info(f"Queued {len(job_ids)} DataStorageSyncJob(s) for project {project.pk}: {job_ids}")
+        return Response({"job_ids": job_ids, "queued": len(job_ids), "project_id": project.pk})
+
     @action(detail=True, methods=["post"], name="regroup-sessions", url_path="regroup-sessions")
     def regroup_sessions(self, _request, pk=None) -> Response:
         """

ami/main/migrations/0095_grant_sync_deployment_to_mldatamanager.py

@@ -0,0 +1,81 @@
+"""
+Grant the existing ``sync_deployment`` permission to ``MLDataManager`` role
+groups on projects that already exist.
+
+``MLDataManager`` already holds ``run_data_storage_sync_job`` (it can run/retry a
+sync job) but not ``sync_deployment`` (which gates *starting* a sync from a
+station via ``POST /api/v2/deployments/<pk>/sync/`` and the bulk
+``.../sync-all/`` action). This closes that gap so ML data managers can trigger
+syncs, not only manage the resulting jobs. ``ProjectManager`` already has the
+permission and is unaffected.
+
+New projects pick this up automatically through ``create_roles_for_project``
+(``MLDataManager`` now includes the permission); this migration backfills the
+role groups of existing projects. The ``sync_deployment`` permission itself is
+already defined on ``Project.Meta.permissions``, so there is no model/schema
+change here — only a data backfill.
+"""
+
+from django.db import migrations
+from django.db.models import Q
+
+
+def grant_sync_to_mldatamanager(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    Permission = apps.get_model("auth", "Permission")
+    ContentType = apps.get_model("contenttypes", "ContentType")
+
+    try:
+        project_ct = ContentType.objects.get(app_label="main", model="project")
+    except ContentType.DoesNotExist:
+        return
+
+    try:
+        perm = Permission.objects.get(codename="sync_deployment", content_type=project_ct)
+    except Permission.DoesNotExist:
+        return
+
+    for group in Group.objects.filter(Q(name__endswith="_MLDataManager")):
+        group.permissions.add(perm)
+
+
+def revoke_sync_from_mldatamanager(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    Permission = apps.get_model("auth", "Permission")
+    ContentType = apps.get_model("contenttypes", "ContentType")
+    GroupObjectPermission = apps.get_model("guardian", "GroupObjectPermission")
+
+    try:
+        project_ct = ContentType.objects.get(app_label="main", model="project")
+    except ContentType.DoesNotExist:
+        return
+
+    try:
+        perm = Permission.objects.get(codename="sync_deployment", content_type=project_ct)
+    except Permission.DoesNotExist:
+        return
+
+    # Only touch MLDataManager groups; ProjectManager holds sync_deployment
+    # independently and must keep it.
+    mldm_groups = Group.objects.filter(Q(name__endswith="_MLDataManager"))
+    for group in mldm_groups:
+        group.permissions.remove(perm)
+    GroupObjectPermission.objects.filter(
+        permission=perm,
+        content_type=project_ct,
+        group__in=mldm_groups,
+    ).delete()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0094_enable_async_pipeline_workers"),
+        ("guardian", "0002_generic_permissions_index"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            grant_sync_to_mldatamanager,
+            revoke_sync_from_mldatamanager,
+        ),
+    ]

ami/main/tests.py

@@ -51,7 +51,14 @@
 )
 from ami.tests.fixtures.storage import populate_bucket
 from ami.users.models import User
-from ami.users.roles import BasicMember, Identifier, MLDataManager, ProjectManager, create_roles_for_project
+from ami.users.roles import (
+    BasicMember,
+    Identifier,
+    MLDataManager,
+    ProjectManager,
+    Researcher,
+    create_roles_for_project,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -2900,6 +2907,105 @@ def test_sync_creates_events_and_updates_counts(self):
         logger.info(f"Initial events count: {initial_events_count}, Updated events count: {updated_events.count()}")
 
 
+class TestDeploymentSyncAll(APITestCase):
+    """
+    The bulk "Sync all" endpoint enqueues one sync job per connected station and
+    is gated by the same sync permission the per-row action enforces.
+
+    Pins three guarantees: the ``data_source_connected`` flag the frontend reads,
+    the permission matrix (a ``detail=False`` action must check permissions
+    itself), and that only stations with a storage source get a job.
+    """
+
+    def setUp(self):
+        super().setUp()
+        from unittest import mock
+
+        self.project = Project.objects.create(name="Sync All Project", description="Sync-all tests")
+        create_roles_for_project(self.project)
+
+        self.superuser = User.objects.create_superuser(email="super-syncall@insectai.org", password="password123")
+        self.pm_user = User.objects.create_user(email="pm-syncall@insectai.org", password="password123")
+        self.ml_user = User.objects.create_user(email="ml-syncall@insectai.org", password="password123")
+        self.researcher = User.objects.create_user(email="researcher-syncall@insectai.org", password="password123")
+        self.identifier = User.objects.create_user(email="identifier-syncall@insectai.org", password="password123")
+        self.basic_user = User.objects.create_user(email="basic-syncall@insectai.org", password="password123")
+        self.outsider = User.objects.create_user(email="outsider-syncall@insectai.org", password="password123")
+        ProjectManager.assign_user(self.pm_user, self.project)
+        MLDataManager.assign_user(self.ml_user, self.project)
+        Researcher.assign_user(self.researcher, self.project)
+        Identifier.assign_user(self.identifier, self.project)
+        BasicMember.assign_user(self.basic_user, self.project)
+
+        source = S3StorageSource.objects.create(
+            name="Sync All Source",
+            bucket="test-bucket",
+            access_key="fake-access-key",
+            secret_key="fake-secret-key",
+            project=self.project,
+        )
+        self.connected = Deployment.objects.create(name="Connected", project=self.project, data_source=source)
+        self.unconnected = Deployment.objects.create(name="Unconnected", project=self.project)
+
+        # Keep the endpoint tests off the broker: assert the job rows and enqueue
+        # calls, not real Celery dispatch.
+        patcher = mock.patch("ami.jobs.models.Job.enqueue")
+        self.mock_enqueue = patcher.start()
+        self.addCleanup(patcher.stop)
+
+        self.url = "/api/v2/deployments/sync-all/"
+
+    def test_data_source_connected_field_in_list(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.get(f"/api/v2/deployments/?project_id={self.project.pk}")
+        self.assertEqual(response.status_code, 200)
+        flags = {row["id"]: row["data_source_connected"] for row in response.data["results"]}
+        self.assertTrue(flags[self.connected.pk], "Connected station should report data_source_connected=True")
+        self.assertFalse(flags[self.unconnected.pk], "Unconnected station should report data_source_connected=False")
+
+    def test_requires_project_id(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.post(self.url)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_permission_matrix(self):
+        # Sync is allowed for MLDataManager and ProjectManager (and superusers),
+        # not for Researcher, Identifier, BasicMember, or non-members.
+        matrix = [
+            ("superuser", self.superuser, status.HTTP_200_OK),
+            ("ProjectManager", self.pm_user, status.HTTP_200_OK),
+            ("MLDataManager", self.ml_user, status.HTTP_200_OK),
+            ("Researcher", self.researcher, status.HTTP_403_FORBIDDEN),
+            ("Identifier", self.identifier, status.HTTP_403_FORBIDDEN),
+            ("BasicMember", self.basic_user, status.HTTP_403_FORBIDDEN),
+            ("outsider", self.outsider, status.HTTP_403_FORBIDDEN),
+        ]
+        for role_name, user, expected in matrix:
+            with self.subTest(role=role_name):
+                self.client.force_authenticate(user)
+                response = self.client.post(f"{self.url}?project_id={self.project.pk}")
+                self.assertEqual(response.status_code, expected, f"{role_name} got {response.status_code}")
+
+    def test_anonymous_denied(self):
+        self.client.force_authenticate(None)
+        response = self.client.post(f"{self.url}?project_id={self.project.pk}")
+        self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
+
+    def test_enqueues_one_job_per_connected_station(self):
+        from ami.jobs.models import DataStorageSyncJob
+
+        self.client.force_authenticate(self.pm_user)
+        response = self.client.post(f"{self.url}?project_id={self.project.pk}")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["queued"], 1)
+        self.assertEqual(len(response.data["job_ids"]), 1)
+        self.assertEqual(self.mock_enqueue.call_count, 1, "One job should be enqueued")
+
+        jobs = Job.objects.filter(project=self.project, job_type_key=DataStorageSyncJob.key)
+        self.assertEqual(jobs.count(), 1, "Exactly one sync job, for the connected station")
+        self.assertEqual(jobs.first().deployment_id, self.connected.pk, "Unconnected station must be skipped")
+
+
 class TestFineGrainedJobRunPermission(APITestCase):
     def setUp(self):
         super().setUp()

ami/users/roles.py

@@ -141,6 +141,9 @@ class MLDataManager(Role):
         Project.Permissions.RUN_ML_JOB,
         Project.Permissions.RUN_POPULATE_CAPTURES_COLLECTION_JOB,
         Project.Permissions.RUN_DATA_STORAGE_SYNC_JOB,
+        # Can start a storage sync from a station, not only run/retry the job it
+        # creates. ProjectManager also has this (inherits MLDataManager).
+        Project.Permissions.SYNC_DEPLOYMENT,
         Project.Permissions.RUN_REGROUP_EVENTS_JOB,
         Project.Permissions.RUN_DATA_EXPORT_JOB,
         Project.Permissions.DELETE_OCCURRENCES,