Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 69 additions & 9 deletions plane/api/work_items/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,15 +4,21 @@
from collections.abc import Mapping
from typing import Any

from ...models.query_params import RetrieveQueryParams, WorkItemQueryParams
from ...models.query_params import (
RetrieveQueryParams,
WorkItemCountQueryParams,
WorkItemQueryParams,
)
from ...models.work_items import (
AdvancedSearchResult,
AdvancedSearchWorkItem,
CreateWorkItem,
PaginatedWorkItemResponse,
UpdateWorkItem,
WorkItem,
WorkItemCountResponse,
WorkItemDetail,
WorkItemGroupedCountResponse,
WorkItemSearch,
)
from ..base_resource import BaseResource
Expand Down Expand Up @@ -47,6 +53,21 @@ def prepare_work_item_params(
return payload


def prepare_work_item_count_params(
params: WorkItemCountQueryParams | None,
) -> dict[str, Any] | None:
"""Serialize work-item count query params for use as HTTP query params.

Same ``filters`` JSON-encoding logic as :func:`prepare_work_item_params`.
"""
if params is None:
return None
payload: dict[str, Any] = params.model_dump(exclude_none=True)
if "filters" in payload and isinstance(payload["filters"], dict):
payload["filters"] = json.dumps(payload["filters"], separators=(",", ":"))
return payload


class WorkItems(BaseResource):
def __init__(self, config: Any) -> None:
super().__init__(config, "/workspaces/")
Expand Down Expand Up @@ -245,22 +266,61 @@ def list_workspace(
)
return PaginatedWorkItemResponse.model_validate(response)

def list_workspace(
def count_workspace(
self,
workspace_slug: str,
params: WorkItemQueryParams | None = None,
) -> PaginatedWorkItemResponse:
"""List work items across the entire workspace.
params: WorkItemCountQueryParams | None = None,
) -> WorkItemCountResponse:
"""Return the count of work items across an entire workspace.

Always returns :class:`WorkItemGroupedCountResponse` with fields
``grouped_by``, ``total_count``, and ``grouped_counts``.

``grouped_counts`` keys are raw ORM field values: UUID strings for
FK/M2M dimensions, plain strings for ``priority`` / ``state__group``,
ISO-date strings for ``target_date`` / ``start_date``. ``"None"`` is
used for work items with no value in that dimension.

Args:
workspace_slug: The workspace slug identifier
params: Optional query parameters for filtering, ordering, and pagination
params: Optional query parameters — supports ``filters``, ``pql``,
and ``group_by``.

Example::

from plane.models.query_params import WorkItemCountQueryParams

# Total count
result = client.work_items.count_workspace(
"my-workspace",
params=WorkItemCountQueryParams(
filters={"priority__in": ["urgent", "high"]},
),
)
print(result.total_count) # e.g. 12

# Grouped by priority
result = client.work_items.count_workspace(
"my-workspace",
params=WorkItemCountQueryParams(group_by="priority"),
)
for group, entry in result.grouped_counts.items():
print(f"{group}: {entry.count}")

# Grouped by state, filtered by PQL
result = client.work_items.count_workspace(
"my-workspace",
params=WorkItemCountQueryParams(
pql='assignee = currentUser()',
group_by="state_id",
),
)
"""
query_params = params.model_dump(exclude_none=True) if params else None
response = self._get(
f"{workspace_slug}/work-items", params=query_params
f"{workspace_slug}/work-items/count",
params=prepare_work_item_count_params(params),
)
return PaginatedWorkItemResponse.model_validate(response)
return WorkItemGroupedCountResponse.model_validate(response)

def search(
self,
Expand Down
59 changes: 58 additions & 1 deletion plane/models/query_params.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Query parameter DTOs for list/retrieve endpoints."""

from typing import Any
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field

Expand Down Expand Up @@ -90,9 +90,66 @@ class RetrieveQueryParams(BaseQueryParams):
model_config = ConfigDict(extra="ignore", populate_by_name=True)


WorkItemCountGroupBy = Literal[
"state_id",
"state__group",
"priority",
"project_id",
"type_id",
"labels__id",
"assignees__id",
"issue_module__module_id",
"release_work_items__release_id",
"cycle_id",
"milestone_id",
"created_by",
"target_date",
"start_date",
]


class WorkItemCountQueryParams(BaseModel):
"""Query parameters for the workspace work item count endpoint.

Accepts the same ``filters`` and ``pql`` as :class:`WorkItemQueryParams`
plus an optional ``group_by`` field.

Without ``group_by`` the response is ``{"count": N}``.
With ``group_by`` the response is
``{"grouped_by": ..., "total_count": N, "results": {group_key: {"count": N}}}``.
"""

model_config = ConfigDict(extra="ignore", populate_by_name=True)

pql: str | None = Field(
None,
description=(
"Plane Query Language expression. Human-readable alternative to "
'`filters`. Example: `priority = "urgent" AND assignee = currentUser()`.'
),
)
filters: dict[str, Any] | None = Field(
None,
description=(
"Structured filter expression. JSON-encoded into the `filters=` "
"query param by the client."
),
)
group_by: WorkItemCountGroupBy | None = Field(
None,
description=(
"ORM field to group counts by. When supplied the response shape "
"changes from a flat ``{count}`` to a grouped "
"``{grouped_by, total_count, results}`` envelope."
),
)


__all__ = [
"BaseQueryParams",
"PaginatedQueryParams",
"RetrieveQueryParams",
"WorkItemCountGroupBy",
"WorkItemCountQueryParams",
"WorkItemQueryParams",
]
40 changes: 39 additions & 1 deletion plane/models/work_items.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from typing import TYPE_CHECKING, Any

from pydantic import BaseModel, ConfigDict, Field
from pydantic import BaseModel, ConfigDict, Field, model_validator

from .enums import AccessEnum, PriorityEnum, WorkItemRelationTypeEnum
from .labels import Label
Expand Down Expand Up @@ -594,3 +594,41 @@ class PaginatedWorkItemLinkResponse(PaginatedResponse):
model_config = ConfigDict(extra="allow", populate_by_name=True)

results: list[WorkItemLink]


class WorkItemGroupCountEntry(BaseModel):
"""Count for a single group in a grouped count response."""

model_config = ConfigDict(extra="allow", populate_by_name=True)

count: int


class WorkItemGroupedCountResponse(BaseModel):
"""Response from the workspace work item count endpoint.

Handles both response shapes from ``GET /workspaces/<slug>/work-items/count``:

**Grouped** (with ``group_by`` param)::

{
"grouped_by": "priority",
"total_count": 42,
"results": {"urgent": {"count": 3}, "none": {"count": 6}}
}

``results`` keys are raw ORM field values: UUID strings for FK/M2M
dimensions, plain strings for ``priority`` / ``state__group``, and
ISO-date strings for ``target_date`` / ``start_date``. The special
key ``"None"`` represents work items with no value in that dimension.
"""

model_config = ConfigDict(extra="allow", populate_by_name=True)

# grouped response
grouped_by: str | None = None
total_count: int | None = None
grouped_counts: dict[str, WorkItemGroupCountEntry] | None = None


WorkItemCountResponse = WorkItemGroupedCountResponse