client.py

# 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.pydantic_utilities import pydantic_v1
from ...core.request_options import RequestOptions
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 .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 IngestionClient:
    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,
    ) -> 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
        -------
        IngestionResponse

        Examples
        --------
        import datetime

        from langfuse import IngestionEvent_TraceCreate, TraceBody
        from langfuse.client import FernLangfuse

        client = FernLangfuse(
            x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME",
            x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION",
            x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
            base_url="https://yourhost.com/path/to/api",
        )
        client.ingestion.batch(
            batch=[
                IngestionEvent_TraceCreate(
                    id="abcdef-1234-5678-90ab",
                    timestamp="2022-01-01T00:00:00.000Z",
                    body=TraceBody(
                        id="abcdef-1234-5678-90ab",
                        timestamp=datetime.datetime.fromisoformat(
                            "2022-01-01 00:00:00+00:00",
                        ),
                        environment="production",
                        name="My Trace",
                        user_id="1234-5678-90ab-cdef",
                        input="My input",
                        output="My output",
                        session_id="1234-5678-90ab-cdef",
                        release="1.0.0",
                        version="1.0.0",
                        metadata="My metadata",
                        tags=["tag1", "tag2"],
                        public=True,
                    ),
                )
            ],
        )
        """
        _response = self._client_wrapper.httpx_client.request(
            "api/public/ingestion",
            method="POST",
            json={"batch": batch, "metadata": metadata},
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(IngestionResponse, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise Error(pydantic_v1.parse_obj_as(typing.Any, _response.json()))  # type: ignore
            if _response.status_code == 401:
                raise UnauthorizedError(
                    pydantic_v1.parse_obj_as(typing.Any, _response.json())
                )  # type: ignore
            if _response.status_code == 403:
                raise AccessDeniedError(
                    pydantic_v1.parse_obj_as(typing.Any, _response.json())
                )  # type: ignore
            if _response.status_code == 405:
                raise MethodNotAllowedError(
                    pydantic_v1.parse_obj_as(typing.Any, _response.json())
                )  # type: ignore
            if _response.status_code == 404:
                raise NotFoundError(
                    pydantic_v1.parse_obj_as(typing.Any, _response.json())
                )  # type: ignore
            _response_json = _response.json()
        except JSONDecodeError:
            raise ApiError(status_code=_response.status_code, body=_response.text)
        raise ApiError(status_code=_response.status_code, body=_response_json)


class AsyncIngestionClient:
    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,
    ) -> 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
        -------
        IngestionResponse

        Examples
        --------
        import asyncio
        import datetime

        from langfuse import IngestionEvent_TraceCreate, TraceBody
        from langfuse.client import AsyncFernLangfuse

        client = AsyncFernLangfuse(
            x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME",
            x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION",
            x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY",
            username="YOUR_USERNAME",
            password="YOUR_PASSWORD",
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.ingestion.batch(
                batch=[
                    IngestionEvent_TraceCreate(
                        id="abcdef-1234-5678-90ab",
                        timestamp="2022-01-01T00:00:00.000Z",
                        body=TraceBody(
                            id="abcdef-1234-5678-90ab",
                            timestamp=datetime.datetime.fromisoformat(
                                "2022-01-01 00:00:00+00:00",
                            ),
                            environment="production",
                            name="My Trace",
                            user_id="1234-5678-90ab-cdef",
                            input="My input",
                            output="My output",
                            session_id="1234-5678-90ab-cdef",
                            release="1.0.0",
                            version="1.0.0",
                            metadata="My metadata",
                            tags=["tag1", "tag2"],
                            public=True,
                        ),
                    )
                ],
            )


        asyncio.run(main())
        """
        _response = await self._client_wrapper.httpx_client.request(
            "api/public/ingestion",
            method="POST",
            json={"batch": batch, "metadata": metadata},
            request_options=request_options,
            omit=OMIT,
        )
        try:
            if 200 <= _response.status_code < 300:
                return pydantic_v1.parse_obj_as(IngestionResponse, _response.json())  # type: ignore
            if _response.status_code == 400:
                raise Error(pydantic_v1.parse_obj_as(typing.Any, _response.json()))  # type: ignore
            if _response.status_code == 401:
                raise UnauthorizedError(
                    pydantic_v1.parse_obj_as(typing.Any, _response.json())
                )  # type: ignore
            if _response.status_code == 403:
                raise AccessDeniedError(
                    pydantic_v1.parse_obj_as(typing.Any, _response.json())
                )  # type: ignore
            if _response.status_code == 405:
                raise MethodNotAllowedError(
                    pydantic_v1.parse_obj_as(typing.Any, _response.json())
                )  # type: ignore
            if _response.status_code == 404:
                raise NotFoundError(
                    pydantic_v1.parse_obj_as(typing.Any, _response.json())
                )  # type: ignore
            _response_json = _response.json()
        except JSONDecodeError:
            raise ApiError(status_code=_response.status_code, body=_response.text)
        raise ApiError(status_code=_response.status_code, body=_response_json)