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,114 @@
+"""
+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.
+
+Permissions here are guardian **object-level** grants on each project, which is
+what ``get_perms(user, project)`` and ``user.has_perm("sync_deployment",
+project)`` read. Adding the permission only to ``group.permissions`` (a global
+Django permission) is not enough — it would not appear in ``get_perms`` and the
+Sync buttons / endpoint would stay inaccessible. So this migration mirrors
+``create_roles_for_project``: it adds the global permission for parity and, more
+importantly, creates the object-level ``GroupObjectPermission`` row per project.
+
+New projects pick this up automatically through ``create_roles_for_project``
+(``MLDataManager`` now includes the permission). A ``post_migrate`` signal
+(``ami.main.apps`` → ``create_roles``) also re-syncs every project's role
+permissions on migrate, so this backfill is belt-and-suspenders; it is kept so
+the grant is explicit and self-contained rather than relying on that signal.
+
+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 _sync_permission(apps):
+    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 None, None
+    try:
+        return Permission.objects.get(codename="sync_deployment", content_type=project_ct), project_ct
+    except Permission.DoesNotExist:
+        return None, None
+
+
+def _project_pk_from_group(group):
+    # Group names are "{project_pk}_{project_name}_{RoleName}"; the pk is the
+    # immutable leading segment (the name can contain underscores).
+    try:
+        return int(group.name.split("_", 1)[0])
+    except (ValueError, IndexError):
+        return None
+
+
+def grant_sync_to_mldatamanager(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    GroupObjectPermission = apps.get_model("guardian", "GroupObjectPermission")
+
+    perm, project_ct = _sync_permission(apps)
+    if perm is None:
+        return
+
+    for group in Group.objects.filter(Q(name__endswith="_MLDataManager")):
+        project_pk = _project_pk_from_group(group)
+        if project_pk is None:
+            continue
+        # Global add for parity with create_roles_for_project; the object-level
+        # row below is what get_perms()/has_perm(perm, project) actually read.
+        group.permissions.add(perm)
+        GroupObjectPermission.objects.get_or_create(
+            permission=perm,
+            content_type=project_ct,
+            object_pk=str(project_pk),
+            group=group,
+        )
+
+
+def revoke_sync_from_mldatamanager(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    GroupObjectPermission = apps.get_model("guardian", "GroupObjectPermission")
+
+    perm, project_ct = _sync_permission(apps)
+    if perm is None:
+        return
+
+    # Only touch MLDataManager groups; ProjectManager holds sync_deployment
+    # independently and must keep it.
+    for group in Group.objects.filter(Q(name__endswith="_MLDataManager")):
+        project_pk = _project_pk_from_group(group)
+        if project_pk is None:
+            continue
+        group.permissions.remove(perm)
+        GroupObjectPermission.objects.filter(
+            permission=perm,
+            content_type=project_ct,
+            object_pk=str(project_pk),
+            group=group,
+        ).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,
+        ),
+    ]