client.py

# This file was auto-generated by Fern from our API Definition.

import typing

from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ..core.request_options import RequestOptions
from ..types.metadata_patch_dto import MetadataPatchDto
from ..types.task_detail_response import TaskDetailResponse
from ..types.task_events_response import TaskEventsResponse
from ..types.task_list_response import TaskListResponse
from ..types.task_response import TaskResponse
from .raw_client import AsyncRawTasksClient, RawTasksClient

# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)


class TasksClient:
    def __init__(self, *, client_wrapper: SyncClientWrapper):
        self._raw_client = RawTasksClient(client_wrapper=client_wrapper)

    @property
    def with_raw_response(self) -> RawTasksClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        RawTasksClient
        """
        return self._raw_client

    def list(
        self,
        *,
        project_id: typing.Optional[str] = None,
        name: typing.Optional[str] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> TaskListResponse:
        """
        Optionally filter by projectId and/or a case-insensitive substring of the task name.

        Parameters
        ----------
        project_id : typing.Optional[str]

        name : typing.Optional[str]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        TaskListResponse
            OK

        Examples
        --------
        from pumpup import PumpUp

        client = PumpUp(
            "0",
            api_key="YOUR_API_KEY",
        )
        client.tasks.list()
        """
        _response = self._raw_client.list(project_id=project_id, name=name, request_options=request_options)
        return _response.data

    def create(
        self,
        *,
        idempotency_key: str,
        name: str,
        project_name: str,
        external_id: typing.Optional[str] = OMIT,
        metadata_patch: typing.Optional[MetadataPatchDto] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> TaskResponse:
        """
        The task opens in the project's initial state; use the returned id as taskId on subsequent events. The name is immutable.

        Parameters
        ----------
        idempotency_key : str
            Idempotency key; duplicate submissions return the original event.

        name : str
            Human-readable case name

        project_name : str
            Slug of an existing project

        external_id : typing.Optional[str]
            Optional client correlation id (e.g. agent session/run); bookkeeping only — not unique

        metadata_patch : typing.Optional[MetadataPatchDto]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        TaskResponse
            Created

        Examples
        --------
        from pumpup import PumpUp

        client = PumpUp(
            "0",
            api_key="YOUR_API_KEY",
        )
        client.tasks.create(
            idempotency_key="Idempotency-Key",
            name="Claim C-1029",
            project_name="claims_refund_v1",
        )
        """
        _response = self._raw_client.create(
            idempotency_key=idempotency_key,
            name=name,
            project_name=project_name,
            external_id=external_id,
            metadata_patch=metadata_patch,
            request_options=request_options,
        )
        return _response.data

    def get(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> TaskDetailResponse:
        """
        Materialized state — name, current state, metadata, attachments — plus open requests awaiting a human.

        Parameters
        ----------
        id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        TaskDetailResponse
            OK

        Examples
        --------
        from pumpup import PumpUp

        client = PumpUp(
            "0",
            api_key="YOUR_API_KEY",
        )
        client.tasks.get(
            id="id",
        )
        """
        _response = self._raw_client.get(id, request_options=request_options)
        return _response.data

    def events(
        self,
        id: str,
        *,
        cursor: typing.Optional[str] = None,
        limit: typing.Optional[int] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> TaskEventsResponse:
        """
        Ascending cursor tail over the task's event log; pass nextCursor back as ?cursor= to resume. An absent cursor starts at the first event.

        Parameters
        ----------
        id : str

        cursor : typing.Optional[str]

        limit : typing.Optional[int]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        TaskEventsResponse
            OK

        Examples
        --------
        from pumpup import PumpUp

        client = PumpUp(
            "0",
            api_key="YOUR_API_KEY",
        )
        client.tasks.events(
            id="id",
        )
        """
        _response = self._raw_client.events(id, cursor=cursor, limit=limit, request_options=request_options)
        return _response.data


class AsyncTasksClient:
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
        self._raw_client = AsyncRawTasksClient(client_wrapper=client_wrapper)

    @property
    def with_raw_response(self) -> AsyncRawTasksClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        AsyncRawTasksClient
        """
        return self._raw_client

    async def list(
        self,
        *,
        project_id: typing.Optional[str] = None,
        name: typing.Optional[str] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> TaskListResponse:
        """
        Optionally filter by projectId and/or a case-insensitive substring of the task name.

        Parameters
        ----------
        project_id : typing.Optional[str]

        name : typing.Optional[str]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        TaskListResponse
            OK

        Examples
        --------
        import asyncio

        from pumpup import AsyncPumpUp

        client = AsyncPumpUp(
            "0",
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.tasks.list()


        asyncio.run(main())
        """
        _response = await self._raw_client.list(project_id=project_id, name=name, request_options=request_options)
        return _response.data

    async def create(
        self,
        *,
        idempotency_key: str,
        name: str,
        project_name: str,
        external_id: typing.Optional[str] = OMIT,
        metadata_patch: typing.Optional[MetadataPatchDto] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> TaskResponse:
        """
        The task opens in the project's initial state; use the returned id as taskId on subsequent events. The name is immutable.

        Parameters
        ----------
        idempotency_key : str
            Idempotency key; duplicate submissions return the original event.

        name : str
            Human-readable case name

        project_name : str
            Slug of an existing project

        external_id : typing.Optional[str]
            Optional client correlation id (e.g. agent session/run); bookkeeping only — not unique

        metadata_patch : typing.Optional[MetadataPatchDto]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        TaskResponse
            Created

        Examples
        --------
        import asyncio

        from pumpup import AsyncPumpUp

        client = AsyncPumpUp(
            "0",
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.tasks.create(
                idempotency_key="Idempotency-Key",
                name="Claim C-1029",
                project_name="claims_refund_v1",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.create(
            idempotency_key=idempotency_key,
            name=name,
            project_name=project_name,
            external_id=external_id,
            metadata_patch=metadata_patch,
            request_options=request_options,
        )
        return _response.data

    async def get(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> TaskDetailResponse:
        """
        Materialized state — name, current state, metadata, attachments — plus open requests awaiting a human.

        Parameters
        ----------
        id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        TaskDetailResponse
            OK

        Examples
        --------
        import asyncio

        from pumpup import AsyncPumpUp

        client = AsyncPumpUp(
            "0",
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.tasks.get(
                id="id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.get(id, request_options=request_options)
        return _response.data

    async def events(
        self,
        id: str,
        *,
        cursor: typing.Optional[str] = None,
        limit: typing.Optional[int] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> TaskEventsResponse:
        """
        Ascending cursor tail over the task's event log; pass nextCursor back as ?cursor= to resume. An absent cursor starts at the first event.

        Parameters
        ----------
        id : str

        cursor : typing.Optional[str]

        limit : typing.Optional[int]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        TaskEventsResponse
            OK

        Examples
        --------
        import asyncio

        from pumpup import AsyncPumpUp

        client = AsyncPumpUp(
            "0",
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.tasks.events(
                id="id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.events(id, cursor=cursor, limit=limit, request_options=request_options)
        return _response.data