# 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 from ..commons.errors.error import Error from ..commons.errors.method_not_allowed_error import MethodNotAllowedError from ..commons.errors.not_found_error import NotFoundError from ..commons.errors.unauthorized_error import UnauthorizedError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.http_response import AsyncHttpResponse, HttpResponse 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.ingestion_event import IngestionEvent from .types.ingestion_response import IngestionResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawIngestionClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def batch( self, *, batch: typing.Sequence[IngestionEvent], metadata: typing.Optional[typing.Any] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[IngestionResponse]: """ **Legacy endpoint for batch ingestion for Langfuse Observability.** -> Please use the OpenTelemetry endpoint (`/api/public/otel/v1/traces`). Learn more: https://langfuse.com/integrations/native/opentelemetry Within each batch, there can be multiple events. Each event has a type, an id, a timestamp, metadata and a body. Internally, we refer to this as the "event envelope" as it tells us something about the event but not the trace. We use the event id within this envelope to deduplicate messages to avoid processing the same event twice, i.e. the event id should be unique per request. The event.body.id is the ID of the actual trace and will be used for updates and will be visible within the Langfuse App. I.e. if you want to update a trace, you'd use the same body id, but separate event IDs. Notes: - Introduction to data model: https://langfuse.com/docs/observability/data-model - Batch sizes are limited to 3.5 MB in total. You need to adjust the number of events per batch accordingly. - The API does not return a 4xx status code for input errors. Instead, it responds with a 207 status code, which includes a list of the encountered errors. Parameters ---------- batch : typing.Sequence[IngestionEvent] Batch of tracing events to be ingested. Discriminated by attribute `type`. metadata : typing.Optional[typing.Any] Optional. Metadata field used by the Langfuse SDKs for debugging. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[IngestionResponse] """ _response = self._client_wrapper.httpx_client.request( "api/public/ingestion", method="POST", json={ "batch": convert_and_respect_annotation_metadata( object_=batch, annotation=typing.Sequence[IngestionEvent], direction="write", ), "metadata": metadata, }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( IngestionResponse, parse_obj_as( type_=IngestionResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) 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 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 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 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 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 AsyncRawIngestionClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def batch( self, *, batch: typing.Sequence[IngestionEvent], metadata: typing.Optional[typing.Any] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[IngestionResponse]: """ **Legacy endpoint for batch ingestion for Langfuse Observability.** -> Please use the OpenTelemetry endpoint (`/api/public/otel/v1/traces`). Learn more: https://langfuse.com/integrations/native/opentelemetry Within each batch, there can be multiple events. Each event has a type, an id, a timestamp, metadata and a body. Internally, we refer to this as the "event envelope" as it tells us something about the event but not the trace. We use the event id within this envelope to deduplicate messages to avoid processing the same event twice, i.e. the event id should be unique per request. The event.body.id is the ID of the actual trace and will be used for updates and will be visible within the Langfuse App. I.e. if you want to update a trace, you'd use the same body id, but separate event IDs. Notes: - Introduction to data model: https://langfuse.com/docs/observability/data-model - Batch sizes are limited to 3.5 MB in total. You need to adjust the number of events per batch accordingly. - The API does not return a 4xx status code for input errors. Instead, it responds with a 207 status code, which includes a list of the encountered errors. Parameters ---------- batch : typing.Sequence[IngestionEvent] Batch of tracing events to be ingested. Discriminated by attribute `type`. metadata : typing.Optional[typing.Any] Optional. Metadata field used by the Langfuse SDKs for debugging. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[IngestionResponse] """ _response = await self._client_wrapper.httpx_client.request( "api/public/ingestion", method="POST", json={ "batch": convert_and_respect_annotation_metadata( object_=batch, annotation=typing.Sequence[IngestionEvent], direction="write", ), "metadata": metadata, }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( IngestionResponse, parse_obj_as( type_=IngestionResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) 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 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 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 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 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, )