python(feat): campaigns resource

python/CHANGELOG.md

@@ -6,6 +6,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 ## [Unreleased]
 
 - Added `client.annotations` for data reviews, phases, and review history.
+- Added `client.campaigns` for grouping report results across runs.
 
 ### What's New
 
@@ -29,7 +30,7 @@ config.complex_types_import_mode = McapComplexTypesImportMode.STRING
 Variable-cardinality fields (dynamic and bounded arrays) are typed `BYTES`. As with Parquet, `complex_types_import_mode` on the config decides what each becomes: Arrow IPC bytes, a JSON string under `<name>.json`, both (the default), or neither.
 
 Reading a file locally needs the new `mcap` extra (`pip install sift-stack-py[mcap]`), so both `detect_config` and importing without a config require it.
-=======
+
 #### Credentials from sift-cli profiles
 
 `SiftClient` now reads the same `sift.toml` profiles that `sift-cli --profile` uses. An environment that you configure once for the CLI works from Python with no arguments.

python/lib/sift_client/_internal/low_level_wrappers/campaigns.py

@@ -0,0 +1,278 @@
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any, cast
+
+from sift.campaigns.v1.campaigns_pb2 import (
+    CampaignReport as CampaignReportProto,
+)
+from sift.campaigns.v1.campaigns_pb2 import (
+    CreateCampaignFrom,
+    CreateCampaignResponse,
+    GetCampaignReportSummariesRequest,
+    GetCampaignReportSummariesResponse,
+    GetCampaignRequest,
+    GetCampaignResponse,
+    ListCampaignsRequest,
+    ListCampaignsResponse,
+    UpdateCampaignRequest,
+    UpdateCampaignResponse,
+)
+from sift.campaigns.v1.campaigns_pb2_grpc import CampaignServiceStub
+from sift.common.type.v1.resource_identifier_pb2 import (
+    Ids,
+    NamedResources,
+    Names,
+    ResourceIdentifiers,
+)
+from sift.tags.v1.tags_pb2 import TagRef
+
+from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase
+from sift_client.sift_types.campaign import (
+    Campaign,
+    CampaignCreate,
+    CampaignReportSummary,
+    CampaignUpdate,
+    report_ids,
+    tag_names,
+)
+from sift_client.transport import WithGrpcClient
+
+if TYPE_CHECKING:
+    from sift_client.transport.grpc_transport import GrpcClient
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+
+class CampaignsLowLevelClient(LowLevelClientBase, WithGrpcClient):
+    """Low-level client for the CampaignsAPI.
+
+    A thin wrapper around the autogenerated bindings for the CampaignService.
+    """
+
+    def __init__(self, grpc_client: GrpcClient):
+        """Initialize the CampaignsLowLevelClient.
+
+        Args:
+            grpc_client: The gRPC client to use for making API calls.
+        """
+        super().__init__(grpc_client)
+
+    async def get_campaign(
+        self,
+        *,
+        campaign_id: str | None = None,
+        client_key: str | None = None,
+        organization_id: str | None = None,
+        skip_report_summaries: bool = False,
+    ) -> Campaign:
+        """Get a campaign by ID or client key.
+
+        Args:
+            campaign_id: The campaign ID to get.
+            client_key: The client key to get, as an alternative to the ID.
+            organization_id: Required with `client_key` if you belong to several orgs.
+            skip_report_summaries: Omit the per-report counts. Much faster for large campaigns.
+
+        Returns:
+            The Campaign.
+
+        Raises:
+            ValueError: If neither identifier is provided.
+        """
+        if campaign_id is None and client_key is None:
+            raise ValueError("Either campaign_id or client_key must be provided")
+
+        request_kwargs: dict[str, Any] = {"skip_report_summaries": skip_report_summaries}
+        if campaign_id is not None:
+            request_kwargs["campaign_id"] = campaign_id
+        if client_key is not None:
+            request_kwargs["client_key"] = client_key
+        if organization_id is not None:
+            request_kwargs["organization_id"] = organization_id
+
+        request = GetCampaignRequest(**request_kwargs)
+        response = await self._grpc_client.get_stub(CampaignServiceStub).GetCampaign(request)
+        return Campaign._from_proto(cast("GetCampaignResponse", response).campaign)
+
+    async def list_campaigns(
+        self,
+        *,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+        page_token: str | None = None,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+        include_archived: bool = False,
+        skip_report_summaries: bool = False,
+    ) -> tuple[list[Campaign], str]:
+        """List one page of campaigns.
+
+        Args:
+            page_size: The maximum number of campaigns to return.
+            page_token: A page token for pagination.
+            query_filter: A CEL filter string.
+            order_by: How to order the retrieved campaigns.
+            include_archived: Include archived campaigns in the results.
+            skip_report_summaries: Omit the per-report counts. Much faster for large campaigns.
+
+        Returns:
+            A tuple of (campaigns, next_page_token).
+        """
+        request_kwargs: dict[str, Any] = {
+            "include_archived": include_archived,
+            "skip_report_summaries": skip_report_summaries,
+        }
+        if page_size is not None:
+            request_kwargs["page_size"] = page_size
+        if page_token is not None:
+            request_kwargs["page_token"] = page_token
+        if query_filter is not None:
+            request_kwargs["filter"] = query_filter
+        if order_by is not None:
+            request_kwargs["order_by"] = order_by
+
+        request = ListCampaignsRequest(**request_kwargs)
+        response = await self._grpc_client.get_stub(CampaignServiceStub).ListCampaigns(request)
+        response = cast("ListCampaignsResponse", response)
+
+        campaigns = [Campaign._from_proto(c) for c in response.campaigns]
+        return campaigns, response.next_page_token
+
+    async def list_all_campaigns(
+        self,
+        *,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+        max_results: int | None = None,
+        include_archived: bool = False,
+        skip_report_summaries: bool = False,
+    ) -> list[Campaign]:
+        """List every matching campaign.
+
+        Args:
+            query_filter: A CEL filter string.
+            order_by: How to order the retrieved campaigns.
+            page_size: The number of results to fetch per request.
+            max_results: Maximum number of results to return.
+            include_archived: Include archived campaigns in the results.
+            skip_report_summaries: Omit the per-report counts. Much faster for large campaigns.
+
+        Returns:
+            A list of all matching campaigns.
+        """
+        return await self._handle_pagination(
+            self.list_campaigns,
+            kwargs={
+                "query_filter": query_filter,
+                "include_archived": include_archived,
+                "skip_report_summaries": skip_report_summaries,
+            },
+            order_by=order_by,
+            max_results=max_results,
+            page_size=page_size,
+        )
+
+    async def create_campaign(
+        self,
+        *,
+        create: CampaignCreate,
+        from_report_ids: list[str] | None = None,
+        from_run_ids: list[str] | None = None,
+        from_campaign_id: str | None = None,
+    ) -> Campaign:
+        """Create a campaign, optionally seeded with reports.
+
+        At most one seed may be given. `from_run_ids` lets the server collect the
+        reports those runs generated.
+
+        Args:
+            create: The campaign definition.
+            from_report_ids: Seed with these reports.
+            from_run_ids: Seed with the reports these runs generated.
+            from_campaign_id: Duplicate this campaign.
+
+        Returns:
+            The created Campaign.
+
+        Raises:
+            ValueError: If more than one seed is provided.
+        """
+        seeds = [s for s in (from_report_ids, from_run_ids, from_campaign_id) if s]
+        if len(seeds) > 1:
+            raise ValueError(
+                "At most one of from_report_ids, from_run_ids, or from_campaign_id may be provided"
+            )
+
+        request_proto = create.to_proto()
+        if create.tags:
+            request_proto.tags.CopyFrom(
+                NamedResources(names=Names(names=tag_names(create.tags)))  # type: ignore[arg-type]
+            )
+        if from_report_ids:
+            request_proto.create_from.CopyFrom(
+                CreateCampaignFrom(reports=ResourceIdentifiers(ids=Ids(ids=from_report_ids)))
+            )
+        elif from_run_ids:
+            request_proto.create_from.CopyFrom(
+                CreateCampaignFrom(runs=ResourceIdentifiers(ids=Ids(ids=from_run_ids)))
+            )
+        elif from_campaign_id:
+            request_proto.create_from.other_campaign.id = from_campaign_id
+
+        response = await self._grpc_client.get_stub(CampaignServiceStub).CreateCampaign(
+            request_proto
+        )
+        return Campaign._from_proto(cast("CreateCampaignResponse", response).campaign)
+
+    async def update_campaign(self, update: CampaignUpdate) -> Campaign:
+        """Update select fields of a campaign.
+
+        Args:
+            update: The updates to apply. Its `resource_id` must be set.
+
+        Returns:
+            The updated Campaign.
+        """
+        grpc_campaign, update_mask = update.to_proto_with_mask()
+        if "reports" in update.model_fields_set:
+            grpc_campaign.reports.extend(
+                CampaignReportProto(report_id=r) for r in report_ids(update.reports)
+            )
+            update_mask.paths.append("reports")
+        if "tags" in update.model_fields_set:
+            grpc_campaign.tags.extend(TagRef(name=name) for name in tag_names(update.tags))
+            update_mask.paths.append("tags")
+        if not update_mask.paths:
+            return await self.get_campaign(campaign_id=grpc_campaign.campaign_id)
+        request = UpdateCampaignRequest(campaign=grpc_campaign, update_mask=update_mask)
+        response = await self._grpc_client.get_stub(CampaignServiceStub).UpdateCampaign(request)
+        return Campaign._from_proto(cast("UpdateCampaignResponse", response).campaign)
+
+    async def get_report_summaries(
+        self, *, campaign_ids: list[str], organization_id: str | None = None
+    ) -> dict[str, list[CampaignReportSummary]]:
+        """Get per-report rule counts for several campaigns at once.
+
+        Args:
+            campaign_ids: The campaigns to summarize.
+            organization_id: Required if you belong to several organizations.
+
+        Returns:
+            A mapping of campaign ID to its reports, with counts populated.
+        """
+        request_kwargs: dict[str, Any] = {"campaign_ids": campaign_ids}
+        if organization_id is not None:
+            request_kwargs["organization_id"] = organization_id
+
+        request = GetCampaignReportSummariesRequest(**request_kwargs)
+        response = await self._grpc_client.get_stub(CampaignServiceStub).GetCampaignReportSummaries(
+            request
+        )
+        response = cast("GetCampaignReportSummariesResponse", response)
+
+        return {
+            campaign_id: [CampaignReportSummary._from_proto(r) for r in reports.reports]
+            for campaign_id, reports in response.summaries_by_campaign_id.items()
+        }

python/lib/sift_client/_tests/conftest.py

@@ -85,6 +85,7 @@ def mock_client():
     client.runs = MagicMock()
     client.channels = MagicMock()
     client.calculated_channels = MagicMock()
+    client.campaigns = MagicMock()
     client.rules = MagicMock()
     client.access_control = MagicMock()
     client.tags = MagicMock()