Aller au contenu

API Module

The main API module provides the core Forra API client functionality.

ScoutAPI

Source code in forrasdk/api/api.py
class ScoutAPI:
    def __init__(
        self,
        base_url: Optional[str] = None,
        api_access_token: Optional[str] = None,
        retry_config: Optional[RetryConfig] = None,
    ) -> None:
        self._base_url = base_url or resolve_var(
            scout.context, VariableNames.FORRA_API_URL
        )
        if self._base_url is None or self._base_url == "":
            raise ValueError(
                f"{VariableNames.FORRA_API_URL} is not set in FORRA_CONTEXT"
            )

        api_access_token = api_access_token or resolve_var(
            scout.context, VariableNames.FORRA_API_ACCESS_TOKEN
        )
        if api_access_token is None or api_access_token == "":
            raise ValueError(
                f"{VariableNames.FORRA_API_ACCESS_TOKEN} is not set in FORRA_CONTEXT"
            )

        self._headers = {
            "Authorization": f"Bearer {api_access_token}",
            "Content-Type": "application/json",
        }

        # Create retry strategy from config or use default
        if retry_config is not None:
            # Empty dict means no retries
            if not retry_config:
                self._retry_strategy = create_retry_strategy(max_attempts=0)
            else:
                self._retry_strategy = create_retry_strategy(**retry_config)
        else:
            self._retry_strategy = retry_on_api_errors

        # Initialize specialized API clients
        self.chat = ChatAPI(self._base_url, self._headers, self._retry_strategy)
        self.assistants = AssistantsAPI(
            self._base_url, self._headers, self._retry_strategy
        )
        self.automations = AutomationsAPI(
            self._base_url, self._headers, self._retry_strategy
        )
        self.conversations = ConversationsAPI(
            self._base_url, self._headers, self._retry_strategy
        )
        self.utils = UtilsAPI(self._base_url, self._headers, self._retry_strategy)
        self.audio = AudioAPI(self._base_url, self._headers, self._retry_strategy)
        self.image = ImageAPI(self._base_url, self._headers, self._retry_strategy)
        self.protected = ProtectedAPI(
            self._base_url, self._headers, self._retry_strategy
        )
        self.skills = SkillsAPI(self._base_url, self._headers, self._retry_strategy)
        self.memories = MemoriesAPI(self._base_url, self._headers, self._retry_strategy)
        self.vision = VisionAPI(self._base_url, self._headers, self._retry_strategy)
        self.video = VideoAPI(self._base_url, self._headers, self._retry_strategy)
        self.search = SearchAPI(self._base_url, self._headers, self._retry_strategy)
        self.projects = ProjectsAPI(self._base_url, self._headers, self._retry_strategy)
        self.users = UsersAPI(self._base_url, self._headers, self._retry_strategy)
        self.auth = AuthAPI(self._base_url, self._headers, self._retry_strategy)
        self.groups = GroupsAPI(self._base_url, self._headers, self._retry_strategy)
        self.external_services = ExternalServicesAPI(
            self._base_url, self._headers, self._retry_strategy
        )

    # Generic requests
    def get(
        self,
        url: str,
        params: RequestType | dict,
        response_model: Optional[Type[ResponseType]] = None,
    ) -> Response[Any]:
        """Make a GET request to the Scout API.

        Args:
            url: The endpoint URL to make the request to (will be appended to base_url).
            params: Request parameters to include in the query string. Can be a BaseModel
                instance or a dictionary.
            response_model: Optional response model class to validate and deserialize
                the response data.

        Returns:
            Response: A Response object containing the status code and validated data.

        Raises:
            ValueError: If the request parameters are invalid.
            RequestException: If the HTTP request fails.
        """
        request_data = params.model_dump() if isinstance(params, BaseModel) else params

        response, status_code = RequestUtils.get(
            url=f"{self._base_url}{url}",
            headers=self._headers,
            params=request_data,
            retry_strategy=self._retry_strategy,
        )

        validated_data = get_validated_data(response, response_model)

        return Response(status_code=status_code, data=validated_data)

    def put(
        self,
        url: str,
        data: RequestType,
        response_model: Optional[Type[ResponseType]] = None,
    ) -> Response[Any]:
        """Make a PUT request to the Scout API.

        Args:
            url: The endpoint URL to make the request to (will be appended to base_url).
            data: Request data to include in the request body. Can be a BaseModel
                instance or a dictionary.
            response_model: Optional response model class to validate and deserialize
                the response data.

        Returns:
            Response: A Response object containing the status code and validated data.

        Raises:
            ValueError: If the request data is invalid.
            RequestException: If the HTTP request fails.
        """
        request_data = data.model_dump() if isinstance(data, BaseModel) else data

        response, status_code = RequestUtils.put(
            url=f"{self._base_url}{url}",
            headers=self._headers,
            payload=request_data,
            retry_strategy=self._retry_strategy,
        )

        validated_data = get_validated_data(response, response_model)

        return Response(status_code=status_code, data=validated_data)

    def post(
        self,
        url: str,
        data: RequestType | dict | None,
        response_model: Optional[Type[ResponseType]] = None,
        files: Optional[dict] = None,
    ) -> Response[Any]:
        """Make a POST request to the Scout API.

        Args:
            url: The endpoint URL to make the request to (will be appended to base_url).
            data: Request data to include in the request body. Can be a BaseModel
                instance or a dictionary.
            response_model: Optional response model class to validate and deserialize
                the response data.
            files: Optional dictionary of files to upload with the request.

        Returns:
            Response: A Response object containing the status code and validated data.

        Raises:
            ValueError: If the request data is invalid.
            RequestException: If the HTTP request fails.
        """
        request_data = data.model_dump() if isinstance(data, BaseModel) else data

        if files:
            local_headers = self._headers.copy()
            local_headers.pop("Content-Type")

            response, status_code = RequestUtils.post(
                url=f"{self._base_url}{url}",
                headers=local_headers,
                data=request_data,
                files=files,
                retry_strategy=self._retry_strategy,
            )
        else:
            response, status_code = RequestUtils.post(
                url=f"{self._base_url}{url}",
                headers=self._headers,
                json_payload=request_data,
                retry_strategy=self._retry_strategy,
            )

        validated_data = get_validated_data(response, response_model)

        return Response(status_code=status_code, data=validated_data)

    def delete(
        self, url: str, response_model: Optional[Type[ResponseType]] = None
    ) -> Response[Any]:
        """Make a DELETE request to the Scout API.

        Args:
            url: The endpoint URL to make the request to (will be appended to base_url).
            response_model: Optional response model class to validate and deserialize
                the response data.

        Returns:
            Response: A Response object containing the status code and validated data.

        Raises:
            ValueError: If the request parameters are invalid.
            RequestException: If the HTTP request fails.
        """
        response, status_code = RequestUtils.delete(
            url=f"{self._base_url}{url}",
            headers=self._headers,
            retry_strategy=self._retry_strategy,
        )

        validated_data = get_validated_data(response, response_model)

        return Response(status_code=status_code, data=validated_data)

delete(url, response_model=None)

Make a DELETE request to the Scout API.

Parameters:

Name Type Description Default
url str

The endpoint URL to make the request to (will be appended to base_url).

required
response_model Optional[Type[ResponseType]]

Optional response model class to validate and deserialize the response data.

None

Returns:

Name Type Description
Response Response[Any]

A Response object containing the status code and validated data.

Raises:

Type Description
ValueError

If the request parameters are invalid.

RequestException

If the HTTP request fails.

Source code in forrasdk/api/api.py
def delete(
    self, url: str, response_model: Optional[Type[ResponseType]] = None
) -> Response[Any]:
    """Make a DELETE request to the Scout API.

    Args:
        url: The endpoint URL to make the request to (will be appended to base_url).
        response_model: Optional response model class to validate and deserialize
            the response data.

    Returns:
        Response: A Response object containing the status code and validated data.

    Raises:
        ValueError: If the request parameters are invalid.
        RequestException: If the HTTP request fails.
    """
    response, status_code = RequestUtils.delete(
        url=f"{self._base_url}{url}",
        headers=self._headers,
        retry_strategy=self._retry_strategy,
    )

    validated_data = get_validated_data(response, response_model)

    return Response(status_code=status_code, data=validated_data)

get(url, params, response_model=None)

Make a GET request to the Scout API.

Parameters:

Name Type Description Default
url str

The endpoint URL to make the request to (will be appended to base_url).

required
params RequestType | dict

Request parameters to include in the query string. Can be a BaseModel instance or a dictionary.

required
response_model Optional[Type[ResponseType]]

Optional response model class to validate and deserialize the response data.

None

Returns:

Name Type Description
Response Response[Any]

A Response object containing the status code and validated data.

Raises:

Type Description
ValueError

If the request parameters are invalid.

RequestException

If the HTTP request fails.

Source code in forrasdk/api/api.py
def get(
    self,
    url: str,
    params: RequestType | dict,
    response_model: Optional[Type[ResponseType]] = None,
) -> Response[Any]:
    """Make a GET request to the Scout API.

    Args:
        url: The endpoint URL to make the request to (will be appended to base_url).
        params: Request parameters to include in the query string. Can be a BaseModel
            instance or a dictionary.
        response_model: Optional response model class to validate and deserialize
            the response data.

    Returns:
        Response: A Response object containing the status code and validated data.

    Raises:
        ValueError: If the request parameters are invalid.
        RequestException: If the HTTP request fails.
    """
    request_data = params.model_dump() if isinstance(params, BaseModel) else params

    response, status_code = RequestUtils.get(
        url=f"{self._base_url}{url}",
        headers=self._headers,
        params=request_data,
        retry_strategy=self._retry_strategy,
    )

    validated_data = get_validated_data(response, response_model)

    return Response(status_code=status_code, data=validated_data)

post(url, data, response_model=None, files=None)

Make a POST request to the Scout API.

Parameters:

Name Type Description Default
url str

The endpoint URL to make the request to (will be appended to base_url).

required
data RequestType | dict | None

Request data to include in the request body. Can be a BaseModel instance or a dictionary.

required
response_model Optional[Type[ResponseType]]

Optional response model class to validate and deserialize the response data.

None
files Optional[dict]

Optional dictionary of files to upload with the request.

None

Returns:

Name Type Description
Response Response[Any]

A Response object containing the status code and validated data.

Raises:

Type Description
ValueError

If the request data is invalid.

RequestException

If the HTTP request fails.

Source code in forrasdk/api/api.py
def post(
    self,
    url: str,
    data: RequestType | dict | None,
    response_model: Optional[Type[ResponseType]] = None,
    files: Optional[dict] = None,
) -> Response[Any]:
    """Make a POST request to the Scout API.

    Args:
        url: The endpoint URL to make the request to (will be appended to base_url).
        data: Request data to include in the request body. Can be a BaseModel
            instance or a dictionary.
        response_model: Optional response model class to validate and deserialize
            the response data.
        files: Optional dictionary of files to upload with the request.

    Returns:
        Response: A Response object containing the status code and validated data.

    Raises:
        ValueError: If the request data is invalid.
        RequestException: If the HTTP request fails.
    """
    request_data = data.model_dump() if isinstance(data, BaseModel) else data

    if files:
        local_headers = self._headers.copy()
        local_headers.pop("Content-Type")

        response, status_code = RequestUtils.post(
            url=f"{self._base_url}{url}",
            headers=local_headers,
            data=request_data,
            files=files,
            retry_strategy=self._retry_strategy,
        )
    else:
        response, status_code = RequestUtils.post(
            url=f"{self._base_url}{url}",
            headers=self._headers,
            json_payload=request_data,
            retry_strategy=self._retry_strategy,
        )

    validated_data = get_validated_data(response, response_model)

    return Response(status_code=status_code, data=validated_data)

put(url, data, response_model=None)

Make a PUT request to the Scout API.

Parameters:

Name Type Description Default
url str

The endpoint URL to make the request to (will be appended to base_url).

required
data RequestType

Request data to include in the request body. Can be a BaseModel instance or a dictionary.

required
response_model Optional[Type[ResponseType]]

Optional response model class to validate and deserialize the response data.

None

Returns:

Name Type Description
Response Response[Any]

A Response object containing the status code and validated data.

Raises:

Type Description
ValueError

If the request data is invalid.

RequestException

If the HTTP request fails.

Source code in forrasdk/api/api.py
def put(
    self,
    url: str,
    data: RequestType,
    response_model: Optional[Type[ResponseType]] = None,
) -> Response[Any]:
    """Make a PUT request to the Scout API.

    Args:
        url: The endpoint URL to make the request to (will be appended to base_url).
        data: Request data to include in the request body. Can be a BaseModel
            instance or a dictionary.
        response_model: Optional response model class to validate and deserialize
            the response data.

    Returns:
        Response: A Response object containing the status code and validated data.

    Raises:
        ValueError: If the request data is invalid.
        RequestException: If the HTTP request fails.
    """
    request_data = data.model_dump() if isinstance(data, BaseModel) else data

    response, status_code = RequestUtils.put(
        url=f"{self._base_url}{url}",
        headers=self._headers,
        payload=request_data,
        retry_strategy=self._retry_strategy,
    )

    validated_data = get_validated_data(response, response_model)

    return Response(status_code=status_code, data=validated_data)