prefect.server.api.clients

WorkPoolsOrchestrationClient

Bases: BaseClient

Source code in src/prefect/server/api/clients.py

class WorkPoolsOrchestrationClient(BaseClient):
    async def __aenter__(self) -> Self:
        return self

    async def read_work_pool(self, work_pool_name: str) -> WorkPool:
        """
        Reads information for a given work pool
        Args:
            work_pool_name: The name of the work pool to for which to get
                information.
        Returns:
            Information about the requested work pool.
        """
        try:
            response = await self._http_client.get(f"/work_pools/{work_pool_name}")
            response.raise_for_status()
            return WorkPool.model_validate(response.json())
        except httpx.HTTPStatusError as e:
            if e.response.status_code == status.HTTP_404_NOT_FOUND:
                raise ObjectNotFound(http_exc=e) from e
            else:
                raise

read_work_pool(work_pool_name) async

Reads information for a given work pool Args: work_pool_name: The name of the work pool to for which to get information. Returns: Information about the requested work pool.

Source code in src/prefect/server/api/clients.py

async def read_work_pool(self, work_pool_name: str) -> WorkPool:
    """
    Reads information for a given work pool
    Args:
        work_pool_name: The name of the work pool to for which to get
            information.
    Returns:
        Information about the requested work pool.
    """
    try:
        response = await self._http_client.get(f"/work_pools/{work_pool_name}")
        response.raise_for_status()
        return WorkPool.model_validate(response.json())
    except httpx.HTTPStatusError as e:
        if e.response.status_code == status.HTTP_404_NOT_FOUND:
            raise ObjectNotFound(http_exc=e) from e
        else:
            raise