# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ...commons.errors.access_denied_error import ( AccessDeniedError as commons_errors_access_denied_error_AccessDeniedError, ) from ...commons.errors.error import Error from ...commons.errors.method_not_allowed_error import ( MethodNotAllowedError as commons_errors_method_not_allowed_error_MethodNotAllowedError, ) from ...commons.errors.not_found_error import ( NotFoundError as commons_errors_not_found_error_NotFoundError, ) from ...commons.errors.unauthorized_error import ( UnauthorizedError as commons_errors_unauthorized_error_UnauthorizedError, ) 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 ..commons.types.evaluation_rule_filter import EvaluationRuleFilter from ..commons.types.evaluation_rule_mapping import EvaluationRuleMapping from ..commons.types.evaluation_rule_target import EvaluationRuleTarget from ..errors.errors.access_denied_error import ( AccessDeniedError as unstable_errors_errors_access_denied_error_AccessDeniedError, ) from ..errors.errors.bad_request_error import BadRequestError from ..errors.errors.conflict_error import ConflictError from ..errors.errors.internal_server_error import InternalServerError from ..errors.errors.method_not_allowed_error import ( MethodNotAllowedError as unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError, ) from ..errors.errors.not_found_error import ( NotFoundError as unstable_errors_errors_not_found_error_NotFoundError, ) from ..errors.errors.too_many_requests_error import TooManyRequestsError from ..errors.errors.unauthorized_error import ( UnauthorizedError as unstable_errors_errors_unauthorized_error_UnauthorizedError, ) from ..errors.errors.unprocessable_content_error import UnprocessableContentError from ..errors.types.public_api_error import PublicApiError from .types.create_evaluation_rule_request import CreateEvaluationRuleRequest from .types.delete_evaluation_rule_response import DeleteEvaluationRuleResponse from .types.evaluation_rule import EvaluationRule from .types.evaluation_rule_evaluator_reference import EvaluationRuleEvaluatorReference from .types.evaluation_rules import EvaluationRules from .types.readable_evaluation_rule import ReadableEvaluationRule # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawEvaluationRulesClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def create( self, *, request: CreateEvaluationRuleRequest, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[EvaluationRule]: """ Create an evaluation rule. An evaluation rule defines **what** incoming data should be evaluated and **how prompt variables should be populated** from that data. Use this resource after choosing an evaluator from the evaluator endpoints. Key rules: - `name` must be unique within the project for public evaluation rules - `target` must be `observation` or `experiment` - `evaluator.name` + `evaluator.scope` must identify an existing evaluator family returned by the evaluator endpoints - Langfuse resolves that family to its latest version before saving the evaluation rule - for `target=experiment`, use dataset `id` values from `GET /api/public/v2/datasets` when filtering by `datasetId` - for `llm_as_judge` evaluators, every evaluator prompt variable must be mapped exactly once - for `code` evaluators, Langfuse uses the fixed code runtime mapping; omit `mapping` in create and update requests - for user-provided `llm_as_judge` mappings, `expected_output` and `experiment_item_metadata` are only valid for `target=experiment` - if `enabled=true`, Langfuse validates that the referenced evaluator can currently run - at most 50 evaluation rules can be effectively active in one project at the same time If an evaluation rule with the same `name` already exists in the project, the API returns `409`. In that case, update the existing resource with `PATCH /api/public/unstable/evaluation-rules/{evaluationRuleId}` instead of creating a second one. If enabling this resource would exceed the 50-active limit, the API also returns `409`. In that case, disable or pause another active evaluation rule before enabling a new one. Current scope: - evaluation rules are live-ingestion rules only - they do not trigger historical backfills Recovery guidance: - `400 invalid_filter_value`: fix the filter `column` or `value` using `details.column`, `details.invalidValues`, and `details.allowedValues` - `400 invalid_filter_value` with `details.column=datasetId`: call `GET /api/public/v2/datasets`, then retry with dataset `id` values from that response - `400 missing_variable_mapping`: for `llm_as_judge` evaluators, fetch the evaluator again and make sure every variable in `variables` appears exactly once in `mapping` - `400 duplicate_variable_mapping`: remove repeated mappings for the same variable - `400 invalid_variable_mapping`: for `llm_as_judge`, switch to a valid `source` for the selected `target`, or fix the variable name - `400 invalid_json_path`: remove or correct the `jsonPath` - `422 evaluator_preflight_failed`: the selected evaluator cannot run with the resolved model configuration. Fix the evaluator/default model setup, then retry the create request. Parameters ---------- request : CreateEvaluationRuleRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[EvaluationRule] """ _response = self._client_wrapper.httpx_client.request( "api/public/unstable/evaluation-rules", method="POST", json=convert_and_respect_annotation_metadata( object_=request, annotation=CreateEvaluationRuleRequest, direction="write", ), request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( EvaluationRule, parse_obj_as( type_=EvaluationRule, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 409: raise ConflictError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableContentError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 list( self, *, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[EvaluationRules]: """ List evaluation rules in the authenticated project. This includes legacy `trace` and `dataset` rules so they can be inspected and migrated to v4 rules. Legacy rules are read-only through this API; create, update, and delete continue to support only `observation` and `experiment` rules. Parameters ---------- page : typing.Optional[int] 1-based page number. Defaults to `1`. limit : typing.Optional[int] Maximum number of items per page. Defaults to `50`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[EvaluationRules] """ _response = self._client_wrapper.httpx_client.request( "api/public/unstable/evaluation-rules", method="GET", params={ "page": page, "limit": limit, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( EvaluationRules, parse_obj_as( type_=EvaluationRules, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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( self, evaluation_rule_id: str, *, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[ReadableEvaluationRule]: """ Get one evaluation rule by its identifier. Use this endpoint to inspect the current evaluator, target, mapping, filters, execution timing, and effective runtime status. Legacy `trace` and `dataset` rules are returned for migration and are read-only through this API. Parameters ---------- evaluation_rule_id : str Evaluation rule identifier returned by the evaluation rule endpoints. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ReadableEvaluationRule] """ _response = self._client_wrapper.httpx_client.request( f"api/public/unstable/evaluation-rules/{jsonable_encoder(evaluation_rule_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ReadableEvaluationRule, parse_obj_as( type_=ReadableEvaluationRule, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 update( self, evaluation_rule_id: str, *, name: typing.Optional[str] = OMIT, evaluator: typing.Optional[EvaluationRuleEvaluatorReference] = OMIT, target: typing.Optional[EvaluationRuleTarget] = OMIT, enabled: typing.Optional[bool] = OMIT, sampling: typing.Optional[float] = OMIT, filter: typing.Optional[typing.Sequence[EvaluationRuleFilter]] = OMIT, mapping: typing.Optional[typing.Sequence[EvaluationRuleMapping]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[EvaluationRule]: """ Update an evaluation rule. Typical uses: - enable or disable live execution - switch to another evaluator - adjust sampling - change filters - update LLM-as-judge variable mappings Important behavior: - provide only the fields you want to change - if you provide `evaluator`, Langfuse resolves that evaluator family to its latest version before saving - changing `target`, `filter`, or an LLM-as-judge `mapping` must still produce a valid target-specific configuration - if you change `target` for an LLM-as-judge rule, also send a compatible `filter` and `mapping` in the same request unless the existing ones are still valid for the new target - for `code` evaluator rules, omit `mapping`; Langfuse stores the fixed code runtime mapping automatically - if the resulting config is enabled, Langfuse re-validates that the selected evaluator can run - if the update would move a non-active evaluation rule into the active state and the project already has 50 active evaluation rules, the API returns `409` Recovery guidance: - if an LLM-as-judge update fails with `missing_variable_mapping` or `invalid_variable_mapping` after changing `evaluator` or `target`, resend the request with a complete new `mapping` - if the update fails with `invalid_filter_value` after changing `target`, resend the request with a target-compatible `filter` Parameters ---------- evaluation_rule_id : str Evaluation rule identifier. name : typing.Optional[str] Updated deployment name. evaluator : typing.Optional[EvaluationRuleEvaluatorReference] Updated evaluator family. Langfuse resolves the provided evaluator family to its latest version before saving the rule. A rule's evaluator type cannot be changed: provide `name` and `scope` for an evaluator family of the rule's current type. To use a different evaluator type, create a new rule. target : typing.Optional[EvaluationRuleTarget] Updated target object type. enabled : typing.Optional[bool] Updated desired enabled state. sampling : typing.Optional[float] Updated sampling fraction. filter : typing.Optional[typing.Sequence[EvaluationRuleFilter]] Updated filter list. For `target=experiment`, `column=datasetId` expects dataset `id` values from `GET /api/public/v2/datasets`, not dataset names. mapping : typing.Optional[typing.Sequence[EvaluationRuleMapping]] Updated LLM-as-judge variable mappings. Do not send this field for code evaluator rules. Langfuse stores the fixed code runtime mapping automatically and returns it in the response. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[EvaluationRule] """ _response = self._client_wrapper.httpx_client.request( f"api/public/unstable/evaluation-rules/{jsonable_encoder(evaluation_rule_id)}", method="PATCH", json={ "name": name, "evaluator": convert_and_respect_annotation_metadata( object_=evaluator, annotation=EvaluationRuleEvaluatorReference, direction="write", ), "target": target, "enabled": enabled, "sampling": sampling, "filter": convert_and_respect_annotation_metadata( object_=filter, annotation=typing.Sequence[EvaluationRuleFilter], direction="write", ), "mapping": convert_and_respect_annotation_metadata( object_=mapping, annotation=typing.Sequence[EvaluationRuleMapping], direction="write", ), }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( EvaluationRule, parse_obj_as( type_=EvaluationRule, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableContentError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 delete( self, evaluation_rule_id: str, *, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[DeleteEvaluationRuleResponse]: """ Delete an evaluation rule. This removes the live-ingestion rule only. It does not delete the referenced evaluator. Parameters ---------- evaluation_rule_id : str Evaluation rule identifier. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DeleteEvaluationRuleResponse] """ _response = self._client_wrapper.httpx_client.request( f"api/public/unstable/evaluation-rules/{jsonable_encoder(evaluation_rule_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteEvaluationRuleResponse, parse_obj_as( type_=DeleteEvaluationRuleResponse, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 AsyncRawEvaluationRulesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def create( self, *, request: CreateEvaluationRuleRequest, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[EvaluationRule]: """ Create an evaluation rule. An evaluation rule defines **what** incoming data should be evaluated and **how prompt variables should be populated** from that data. Use this resource after choosing an evaluator from the evaluator endpoints. Key rules: - `name` must be unique within the project for public evaluation rules - `target` must be `observation` or `experiment` - `evaluator.name` + `evaluator.scope` must identify an existing evaluator family returned by the evaluator endpoints - Langfuse resolves that family to its latest version before saving the evaluation rule - for `target=experiment`, use dataset `id` values from `GET /api/public/v2/datasets` when filtering by `datasetId` - for `llm_as_judge` evaluators, every evaluator prompt variable must be mapped exactly once - for `code` evaluators, Langfuse uses the fixed code runtime mapping; omit `mapping` in create and update requests - for user-provided `llm_as_judge` mappings, `expected_output` and `experiment_item_metadata` are only valid for `target=experiment` - if `enabled=true`, Langfuse validates that the referenced evaluator can currently run - at most 50 evaluation rules can be effectively active in one project at the same time If an evaluation rule with the same `name` already exists in the project, the API returns `409`. In that case, update the existing resource with `PATCH /api/public/unstable/evaluation-rules/{evaluationRuleId}` instead of creating a second one. If enabling this resource would exceed the 50-active limit, the API also returns `409`. In that case, disable or pause another active evaluation rule before enabling a new one. Current scope: - evaluation rules are live-ingestion rules only - they do not trigger historical backfills Recovery guidance: - `400 invalid_filter_value`: fix the filter `column` or `value` using `details.column`, `details.invalidValues`, and `details.allowedValues` - `400 invalid_filter_value` with `details.column=datasetId`: call `GET /api/public/v2/datasets`, then retry with dataset `id` values from that response - `400 missing_variable_mapping`: for `llm_as_judge` evaluators, fetch the evaluator again and make sure every variable in `variables` appears exactly once in `mapping` - `400 duplicate_variable_mapping`: remove repeated mappings for the same variable - `400 invalid_variable_mapping`: for `llm_as_judge`, switch to a valid `source` for the selected `target`, or fix the variable name - `400 invalid_json_path`: remove or correct the `jsonPath` - `422 evaluator_preflight_failed`: the selected evaluator cannot run with the resolved model configuration. Fix the evaluator/default model setup, then retry the create request. Parameters ---------- request : CreateEvaluationRuleRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[EvaluationRule] """ _response = await self._client_wrapper.httpx_client.request( "api/public/unstable/evaluation-rules", method="POST", json=convert_and_respect_annotation_metadata( object_=request, annotation=CreateEvaluationRuleRequest, direction="write", ), request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( EvaluationRule, parse_obj_as( type_=EvaluationRule, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 409: raise ConflictError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableContentError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 list( self, *, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[EvaluationRules]: """ List evaluation rules in the authenticated project. This includes legacy `trace` and `dataset` rules so they can be inspected and migrated to v4 rules. Legacy rules are read-only through this API; create, update, and delete continue to support only `observation` and `experiment` rules. Parameters ---------- page : typing.Optional[int] 1-based page number. Defaults to `1`. limit : typing.Optional[int] Maximum number of items per page. Defaults to `50`. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[EvaluationRules] """ _response = await self._client_wrapper.httpx_client.request( "api/public/unstable/evaluation-rules", method="GET", params={ "page": page, "limit": limit, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( EvaluationRules, parse_obj_as( type_=EvaluationRules, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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( self, evaluation_rule_id: str, *, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[ReadableEvaluationRule]: """ Get one evaluation rule by its identifier. Use this endpoint to inspect the current evaluator, target, mapping, filters, execution timing, and effective runtime status. Legacy `trace` and `dataset` rules are returned for migration and are read-only through this API. Parameters ---------- evaluation_rule_id : str Evaluation rule identifier returned by the evaluation rule endpoints. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ReadableEvaluationRule] """ _response = await self._client_wrapper.httpx_client.request( f"api/public/unstable/evaluation-rules/{jsonable_encoder(evaluation_rule_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ReadableEvaluationRule, parse_obj_as( type_=ReadableEvaluationRule, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 update( self, evaluation_rule_id: str, *, name: typing.Optional[str] = OMIT, evaluator: typing.Optional[EvaluationRuleEvaluatorReference] = OMIT, target: typing.Optional[EvaluationRuleTarget] = OMIT, enabled: typing.Optional[bool] = OMIT, sampling: typing.Optional[float] = OMIT, filter: typing.Optional[typing.Sequence[EvaluationRuleFilter]] = OMIT, mapping: typing.Optional[typing.Sequence[EvaluationRuleMapping]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[EvaluationRule]: """ Update an evaluation rule. Typical uses: - enable or disable live execution - switch to another evaluator - adjust sampling - change filters - update LLM-as-judge variable mappings Important behavior: - provide only the fields you want to change - if you provide `evaluator`, Langfuse resolves that evaluator family to its latest version before saving - changing `target`, `filter`, or an LLM-as-judge `mapping` must still produce a valid target-specific configuration - if you change `target` for an LLM-as-judge rule, also send a compatible `filter` and `mapping` in the same request unless the existing ones are still valid for the new target - for `code` evaluator rules, omit `mapping`; Langfuse stores the fixed code runtime mapping automatically - if the resulting config is enabled, Langfuse re-validates that the selected evaluator can run - if the update would move a non-active evaluation rule into the active state and the project already has 50 active evaluation rules, the API returns `409` Recovery guidance: - if an LLM-as-judge update fails with `missing_variable_mapping` or `invalid_variable_mapping` after changing `evaluator` or `target`, resend the request with a complete new `mapping` - if the update fails with `invalid_filter_value` after changing `target`, resend the request with a target-compatible `filter` Parameters ---------- evaluation_rule_id : str Evaluation rule identifier. name : typing.Optional[str] Updated deployment name. evaluator : typing.Optional[EvaluationRuleEvaluatorReference] Updated evaluator family. Langfuse resolves the provided evaluator family to its latest version before saving the rule. A rule's evaluator type cannot be changed: provide `name` and `scope` for an evaluator family of the rule's current type. To use a different evaluator type, create a new rule. target : typing.Optional[EvaluationRuleTarget] Updated target object type. enabled : typing.Optional[bool] Updated desired enabled state. sampling : typing.Optional[float] Updated sampling fraction. filter : typing.Optional[typing.Sequence[EvaluationRuleFilter]] Updated filter list. For `target=experiment`, `column=datasetId` expects dataset `id` values from `GET /api/public/v2/datasets`, not dataset names. mapping : typing.Optional[typing.Sequence[EvaluationRuleMapping]] Updated LLM-as-judge variable mappings. Do not send this field for code evaluator rules. Langfuse stores the fixed code runtime mapping automatically and returns it in the response. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[EvaluationRule] """ _response = await self._client_wrapper.httpx_client.request( f"api/public/unstable/evaluation-rules/{jsonable_encoder(evaluation_rule_id)}", method="PATCH", json={ "name": name, "evaluator": convert_and_respect_annotation_metadata( object_=evaluator, annotation=EvaluationRuleEvaluatorReference, direction="write", ), "target": target, "enabled": enabled, "sampling": sampling, "filter": convert_and_respect_annotation_metadata( object_=filter, annotation=typing.Sequence[EvaluationRuleFilter], direction="write", ), "mapping": convert_and_respect_annotation_metadata( object_=mapping, annotation=typing.Sequence[EvaluationRuleMapping], direction="write", ), }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( EvaluationRule, parse_obj_as( type_=EvaluationRule, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableContentError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 delete( self, evaluation_rule_id: str, *, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[DeleteEvaluationRuleResponse]: """ Delete an evaluation rule. This removes the live-ingestion rule only. It does not delete the referenced evaluator. Parameters ---------- evaluation_rule_id : str Evaluation rule identifier. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DeleteEvaluationRuleResponse] """ _response = await self._client_wrapper.httpx_client.request( f"api/public/unstable/evaluation-rules/{jsonable_encoder(evaluation_rule_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteEvaluationRuleResponse, parse_obj_as( type_=DeleteEvaluationRuleResponse, # 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( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise unstable_errors_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise unstable_errors_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise unstable_errors_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise unstable_errors_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( PublicApiError, parse_obj_as( type_=PublicApiError, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise commons_errors_unauthorized_error_UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise commons_errors_access_denied_error_AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise commons_errors_method_not_allowed_error_MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise commons_errors_not_found_error_NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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, )