# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing from json.decoder import JSONDecodeError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.datetime_utils import serialize_datetime from ..core.http_response import AsyncHttpResponse, HttpResponse from ..core.jsonable_encoder import jsonable_encoder from ..core.request_options import RequestOptions from ..core.unchecked_base_model import construct_type from ..errors.bad_request_error import BadRequestError from ..errors.not_found_error import NotFoundError from ..errors.payment_required_error import PaymentRequiredError from ..errors.unprocessable_entity_error import UnprocessableEntityError from ..types.execute_skill_response import ExecuteSkillResponse from ..types.marketplace_skill_list_response import MarketplaceSkillListResponse from ..types.marketplace_skill_response import MarketplaceSkillResponse from ..types.skill_category import SkillCategory from ..types.skill_response import SkillResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawSkillsMarketplaceClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list_skills( self, *, page_size: typing.Optional[int] = None, page_number: typing.Optional[int] = None, query: typing.Optional[str] = None, category: typing.Optional[SkillCategory] = None, from_date: typing.Optional[dt.datetime] = None, to_date: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[MarketplaceSkillListResponse]: """ List all public skills available in the marketplace with optional filtering. Parameters ---------- page_size : typing.Optional[int] page_number : typing.Optional[int] query : typing.Optional[str] category : typing.Optional[SkillCategory] from_date : typing.Optional[dt.datetime] to_date : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[MarketplaceSkillListResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( "marketplace/skills", method="GET", params={ "pageSize": page_size, "pageNumber": page_number, "query": query, "category": category, "fromDate": serialize_datetime(from_date) if from_date is not None else None, "toDate": serialize_datetime(to_date) if to_date is not None else None, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( MarketplaceSkillListResponse, construct_type( type_=MarketplaceSkillListResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_skill( self, skill_slug: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[MarketplaceSkillResponse]: """ Get details of a specific public skill from the marketplace. Parameters ---------- skill_slug : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[MarketplaceSkillResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"marketplace/skills/{jsonable_encoder(skill_slug)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( MarketplaceSkillResponse, construct_type( type_=MarketplaceSkillResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def clone_skill( self, skill_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[SkillResponse]: """ Clone a public marketplace skill to the user's project. Parameters ---------- skill_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[SkillResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"marketplace/skills/{jsonable_encoder(skill_id)}/clone", method="POST", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SkillResponse, construct_type( type_=SkillResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def execute_skill( self, skill_id: str, *, parameters: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, session_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ExecuteSkillResponse]: """ Execute a skill with the provided parameters. Parameters ---------- skill_id : str parameters : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] Parameters to pass to the skill handler session_id : typing.Optional[str] Optional session ID (UUID) for IP persistence. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ExecuteSkillResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"marketplace/skills/{jsonable_encoder(skill_id)}/execute", method="POST", json={ "parameters": parameters, "sessionId": session_id, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExecuteSkillResponse, construct_type( type_=ExecuteSkillResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 402: raise PaymentRequiredError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawSkillsMarketplaceClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list_skills( self, *, page_size: typing.Optional[int] = None, page_number: typing.Optional[int] = None, query: typing.Optional[str] = None, category: typing.Optional[SkillCategory] = None, from_date: typing.Optional[dt.datetime] = None, to_date: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[MarketplaceSkillListResponse]: """ List all public skills available in the marketplace with optional filtering. Parameters ---------- page_size : typing.Optional[int] page_number : typing.Optional[int] query : typing.Optional[str] category : typing.Optional[SkillCategory] from_date : typing.Optional[dt.datetime] to_date : typing.Optional[dt.datetime] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[MarketplaceSkillListResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "marketplace/skills", method="GET", params={ "pageSize": page_size, "pageNumber": page_number, "query": query, "category": category, "fromDate": serialize_datetime(from_date) if from_date is not None else None, "toDate": serialize_datetime(to_date) if to_date is not None else None, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( MarketplaceSkillListResponse, construct_type( type_=MarketplaceSkillListResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_skill( self, skill_slug: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[MarketplaceSkillResponse]: """ Get details of a specific public skill from the marketplace. Parameters ---------- skill_slug : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[MarketplaceSkillResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"marketplace/skills/{jsonable_encoder(skill_slug)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( MarketplaceSkillResponse, construct_type( type_=MarketplaceSkillResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def clone_skill( self, skill_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[SkillResponse]: """ Clone a public marketplace skill to the user's project. Parameters ---------- skill_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[SkillResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"marketplace/skills/{jsonable_encoder(skill_id)}/clone", method="POST", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SkillResponse, construct_type( type_=SkillResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def execute_skill( self, skill_id: str, *, parameters: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT, session_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ExecuteSkillResponse]: """ Execute a skill with the provided parameters. Parameters ---------- skill_id : str parameters : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] Parameters to pass to the skill handler session_id : typing.Optional[str] Optional session ID (UUID) for IP persistence. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ExecuteSkillResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"marketplace/skills/{jsonable_encoder(skill_id)}/execute", method="POST", json={ "parameters": parameters, "sessionId": session_id, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ExecuteSkillResponse, construct_type( type_=ExecuteSkillResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 402: raise PaymentRequiredError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)