# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.http_response import AsyncHttpResponse, HttpResponse from ..core.jsonable_encoder import jsonable_encoder from ..core.pydantic_utilities import parse_obj_as from ..core.request_options import RequestOptions from ..core.serialization import convert_and_respect_annotation_metadata from ..types.automation_rule_evaluator_page_public import AutomationRuleEvaluatorPagePublic from ..types.automation_rule_evaluator_public import AutomationRuleEvaluatorPublic from ..types.automation_rule_evaluator_update import AutomationRuleEvaluatorUpdate from ..types.automation_rule_evaluator_write import AutomationRuleEvaluatorWrite from ..types.log_page import LogPage # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawAutomationRuleEvaluatorsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def find_evaluators( self, *, project_id: typing.Optional[str] = None, id: typing.Optional[str] = None, name: typing.Optional[str] = None, filters: typing.Optional[str] = None, sorting: typing.Optional[str] = None, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AutomationRuleEvaluatorPagePublic]: """ Find project Evaluators Parameters ---------- project_id : typing.Optional[str] id : typing.Optional[str] name : typing.Optional[str] filters : typing.Optional[str] sorting : typing.Optional[str] page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AutomationRuleEvaluatorPagePublic] Evaluators resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/automations/evaluators", method="GET", params={ "project_id": project_id, "id": id, "name": name, "filters": filters, "sorting": sorting, "page": page, "size": size, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AutomationRuleEvaluatorPagePublic, parse_obj_as( type_=AutomationRuleEvaluatorPagePublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _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 create_automation_rule_evaluator( self, *, request: AutomationRuleEvaluatorWrite, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Create automation rule evaluator Parameters ---------- request : AutomationRuleEvaluatorWrite request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/automations/evaluators", method="POST", json=convert_and_respect_annotation_metadata( object_=request, annotation=AutomationRuleEvaluatorWrite, direction="write" ), headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _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 delete_automation_rule_evaluator_batch( self, *, ids: typing.Sequence[str], project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Delete automation rule evaluators batch Parameters ---------- ids : typing.Sequence[str] project_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/automations/evaluators/delete", method="POST", params={ "project_id": project_id, }, json={ "ids": ids, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _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_evaluator_by_id( self, id: str, *, project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[AutomationRuleEvaluatorPublic]: """ Get automation rule by id Parameters ---------- id : str project_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[AutomationRuleEvaluatorPublic] Automation Rule resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/automations/evaluators/{jsonable_encoder(id)}", method="GET", params={ "project_id": project_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AutomationRuleEvaluatorPublic, parse_obj_as( type_=AutomationRuleEvaluatorPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _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 update_automation_rule_evaluator( self, id: str, *, request: AutomationRuleEvaluatorUpdate, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Update Automation Rule Evaluator by id Parameters ---------- id : str request : AutomationRuleEvaluatorUpdate request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/automations/evaluators/{jsonable_encoder(id)}", method="PATCH", json=convert_and_respect_annotation_metadata( object_=request, annotation=AutomationRuleEvaluatorUpdate, direction="write" ), headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) _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_evaluator_logs_by_id( self, id: str, *, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[LogPage]: """ Get automation rule evaluator logs by id Parameters ---------- id : str size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[LogPage] Automation rule evaluator logs resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/automations/evaluators/{jsonable_encoder(id)}/logs", method="GET", params={ "size": size, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( LogPage, parse_obj_as( type_=LogPage, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) _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 AsyncRawAutomationRuleEvaluatorsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def find_evaluators( self, *, project_id: typing.Optional[str] = None, id: typing.Optional[str] = None, name: typing.Optional[str] = None, filters: typing.Optional[str] = None, sorting: typing.Optional[str] = None, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AutomationRuleEvaluatorPagePublic]: """ Find project Evaluators Parameters ---------- project_id : typing.Optional[str] id : typing.Optional[str] name : typing.Optional[str] filters : typing.Optional[str] sorting : typing.Optional[str] page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AutomationRuleEvaluatorPagePublic] Evaluators resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/automations/evaluators", method="GET", params={ "project_id": project_id, "id": id, "name": name, "filters": filters, "sorting": sorting, "page": page, "size": size, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AutomationRuleEvaluatorPagePublic, parse_obj_as( type_=AutomationRuleEvaluatorPagePublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _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 create_automation_rule_evaluator( self, *, request: AutomationRuleEvaluatorWrite, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Create automation rule evaluator Parameters ---------- request : AutomationRuleEvaluatorWrite request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/automations/evaluators", method="POST", json=convert_and_respect_annotation_metadata( object_=request, annotation=AutomationRuleEvaluatorWrite, direction="write" ), headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _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 delete_automation_rule_evaluator_batch( self, *, ids: typing.Sequence[str], project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Delete automation rule evaluators batch Parameters ---------- ids : typing.Sequence[str] project_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/automations/evaluators/delete", method="POST", params={ "project_id": project_id, }, json={ "ids": ids, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _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_evaluator_by_id( self, id: str, *, project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[AutomationRuleEvaluatorPublic]: """ Get automation rule by id Parameters ---------- id : str project_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[AutomationRuleEvaluatorPublic] Automation Rule resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/automations/evaluators/{jsonable_encoder(id)}", method="GET", params={ "project_id": project_id, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( AutomationRuleEvaluatorPublic, parse_obj_as( type_=AutomationRuleEvaluatorPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _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 update_automation_rule_evaluator( self, id: str, *, request: AutomationRuleEvaluatorUpdate, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Update Automation Rule Evaluator by id Parameters ---------- id : str request : AutomationRuleEvaluatorUpdate request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/automations/evaluators/{jsonable_encoder(id)}", method="PATCH", json=convert_and_respect_annotation_metadata( object_=request, annotation=AutomationRuleEvaluatorUpdate, direction="write" ), headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) _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_evaluator_logs_by_id( self, id: str, *, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[LogPage]: """ Get automation rule evaluator logs by id Parameters ---------- id : str size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[LogPage] Automation rule evaluator logs resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/automations/evaluators/{jsonable_encoder(id)}/logs", method="GET", params={ "size": size, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( LogPage, parse_obj_as( type_=LogPage, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) _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)