client.py

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

import enum
import re
import typing

import httpx
from ..core.client_wrapper import (
    AsyncClientWrapper,
    SyncClientWrapper,
)
from ..core.request_options import RequestOptions

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


class HTTPAuthType(enum.Enum):
    """Authentication types for workflow invocation"""
    NONE = "none"
    STATIC_BEARER = "static_bearer_token"
    OAUTH = "oauth"


class WorkflowsClient:

    def __init__(
        self,
        *,
        client_wrapper: SyncClientWrapper,
        workflow_domain: str = "m.pipedream.net",
    ):
        self._client_wrapper = client_wrapper
        self._workflow_domain = workflow_domain

    def invoke(
        self,
        url_or_endpoint: str,
        *,
        method: str = "POST",
        body: typing.Optional[typing.Any] = None,
        headers: typing.Optional[typing.Dict[str, str]] = None,
        auth_type: HTTPAuthType = HTTPAuthType.NONE,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> httpx.Response:
        """
        Invokes a workflow using its HTTP interface URL by sending an HTTP
        request.

        Parameters
        ----------
        url_or_endpoint : str
            The URL of the workflow's HTTP interface or the ID of the endpoint

        method : str
            HTTP method to use (default: "POST")

        body : typing.Optional[typing.Any]
            Request body data

        headers : typing.Optional[typing.Dict[str, str]]
            HTTP headers to include

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

        Returns
        -------
        httpx.Response
            Response from the workflow
        """
        workflow_url = self._build_workflow_url(url_or_endpoint)
        request_headers = self._prepare_headers(headers, auth_type)

        # Make the request
        response = self._client_wrapper.httpx_client.request(
            workflow_url,
            method=method,
            data=body,
            json=body,
            headers=request_headers,
            request_options=request_options,
        )
        return response.raise_for_status()

    def invoke_for_external_user(
        self,
        url_or_endpoint: str,
        *,
        external_user_id: str,
        method: str = "POST",
        body: typing.Optional[typing.Any] = None,
        headers: typing.Optional[typing.Dict[str, str]] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> httpx.Response:
        """
        Invokes a workflow for a specific Pipedream Connect user in a project.

        Parameters
        ----------
        url_or_endpoint : str
            The URL of the workflow's HTTP interface or the ID of the endpoint

        external_user_id : str
            The external user ID for whom the workflow is being invoked

        method : str
            HTTP method to use (default: "POST")

        body : typing.Optional[typing.Any]
            Request body data

        headers : typing.Optional[typing.Dict[str, str]]
            HTTP headers to include

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

        Returns
        -------
        httpx.Response
            Response from the workflow
        """
        if not external_user_id:
            raise ValueError("external_user_id is required")

        # Add the external user ID header
        request_headers = dict(headers) if headers else {}
        request_headers["X-PD-External-User-ID"] = external_user_id

        # Force OAuth authentication for external user invocations
        return self.invoke(
            url_or_endpoint,
            method=method,
            body=body,
            headers=request_headers,
            auth_type=HTTPAuthType.OAUTH,
            request_options=request_options,
        )

    def _build_workflow_url(self, url_or_endpoint: str) -> str:
        """Build the full workflow URL from either a full URL or endpoint ID"""
        if not url_or_endpoint:
            raise ValueError("URL or endpoint ID is required")

        # Check if it's already a full URL with protocol
        if url_or_endpoint.startswith(("http://", "https://")):
            return url_or_endpoint

        # Check if it's a URL without protocol
        if "." in url_or_endpoint and not self._is_endpoint(url_or_endpoint):
            return f"https://{url_or_endpoint}"

        # Treat as endpoint ID - validate format
        if not self._is_endpoint(url_or_endpoint):
            raise ValueError(f"Invalid endpoint ID format: {url_or_endpoint}")

        # Build URL from endpoint ID
        return f"https://{url_or_endpoint}.{self._workflow_domain}"

    def _is_endpoint(self, url_or_endpoint: str) -> bool:
        """Check if the provided string is an endpoint ID"""
        return bool(re.match(r"^e(n|o)[a-z0-9-]+$", url_or_endpoint))

    def _prepare_headers(
        self,
        headers: typing.Optional[typing.Dict[str, str]],
        auth_type: HTTPAuthType,
    ) -> typing.Dict[str, str]:
        """Prepare headers for the workflow request"""
        request_headers = dict(headers) if headers else {}

        if auth_type == HTTPAuthType.NONE:
            request_headers["Authorization"] = ""

        return request_headers


class AsyncWorkflowsClient:

    def __init__(
        self,
        *,
        client_wrapper: AsyncClientWrapper,
        workflow_domain: str = "m.pipedream.net",
    ):
        self._client_wrapper = client_wrapper
        self._workflow_domain = workflow_domain

    async def invoke(
        self,
        url_or_endpoint: str,
        *,
        method: str = "POST",
        body: typing.Optional[typing.Any] = None,
        headers: typing.Optional[typing.Dict[str, str]] = None,
        auth_type: HTTPAuthType = HTTPAuthType.NONE,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> httpx.Response:
        """
        Invokes a workflow using its HTTP interface URL by sending an HTTP
        request.

        Parameters
        ----------
        url_or_endpoint : str
            The URL of the workflow's HTTP interface or the ID of the endpoint

        method : str
            HTTP method to use (default: "POST")

        body : typing.Optional[typing.Any]
            Request body data

        headers : typing.Optional[typing.Dict[str, str]]
            HTTP headers to include

        auth_type : HTTPAuthType
            Type of authorization (default: HTTPAuthType.NONE)

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

        Returns
        -------
        httpx.Response
            Response from the workflow
        """
        workflow_url = self._build_workflow_url(url_or_endpoint)
        request_headers = self._prepare_headers(headers, auth_type)

        # Make the request
        response = await self._client_wrapper.httpx_client.request(
            workflow_url,
            method=method,
            data=body,
            json=body,
            headers=request_headers,
            request_options=request_options,
        )
        return response.raise_for_status()

    async def invoke_for_external_user(
        self,
        url_or_endpoint: str,
        *,
        external_user_id: str,
        method: str = "POST",
        body: typing.Optional[typing.Any] = None,
        headers: typing.Optional[typing.Dict[str, str]] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> httpx.Response:
        """
        Invokes a workflow for a specific Pipedream Connect user in a project.

        Parameters
        ----------
        url_or_endpoint : str
            The URL of the workflow's HTTP interface or the ID of the endpoint

        external_user_id : str
            The external user ID for whom the workflow is being invoked

        method : str
            HTTP method to use (default: "POST")

        body : typing.Optional[typing.Any]
            Request body data

        headers : typing.Optional[typing.Dict[str, str]]
            HTTP headers to include

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

        Returns
        -------
        httpx.Response
            Response from the workflow
        """
        if not external_user_id:
            raise ValueError("external_user_id is required")

        # Add the external user ID header
        request_headers = dict(headers) if headers else {}
        request_headers["X-PD-External-User-ID"] = external_user_id

        # Force OAuth authentication for external user invocations
        return await self.invoke(
            url_or_endpoint,
            method=method,
            body=body,
            headers=request_headers,
            auth_type=HTTPAuthType.OAUTH,
            request_options=request_options,
        )

    def _build_workflow_url(self, url_or_endpoint: str) -> str:
        """Build the full workflow URL from either a full URL or endpoint ID"""
        if not url_or_endpoint:
            raise ValueError("URL or endpoint ID is required")

        # Check if it's already a full URL with protocol
        if url_or_endpoint.startswith(("http://", "https://")):
            return url_or_endpoint

        # Check if it's a URL without protocol
        if "." in url_or_endpoint and not self._is_endpoint(url_or_endpoint):
            return f"https://{url_or_endpoint}"

        # Treat as endpoint ID - validate format
        if not self._is_endpoint(url_or_endpoint):
            raise ValueError(f"Invalid endpoint ID format: {url_or_endpoint}")

        # Build URL from endpoint ID
        return f"https://{url_or_endpoint}.{self._workflow_domain}"

    def _is_endpoint(self, url_or_endpoint: str) -> bool:
        """Check if the provided string is an endpoint ID"""
        return bool(re.match(r"^e(n|o)[a-z0-9-]+$", url_or_endpoint))

    def _prepare_headers(
        self,
        headers: typing.Optional[typing.Dict[str, str]],
        auth_type: HTTPAuthType,
    ) -> typing.Dict[str, str]:
        """Prepare headers for the workflow request"""
        request_headers = dict(headers) if headers else {}

        if auth_type == HTTPAuthType.NONE:
            request_headers["Authorization"] = ""

        return request_headers