langfuse.request

  1"""@private"""
  2
  3import json
  4import logging
  5from base64 import b64encode
  6from typing import Any, List, Union
  7
  8import httpx
  9
 10from langfuse.serializer import EventSerializer
 11
 12
 13class LangfuseClient:
 14    _public_key: str
 15    _secret_key: str
 16    _base_url: str
 17    _version: str
 18    _timeout: int
 19    _session: httpx.Client
 20
 21    def __init__(
 22        self,
 23        public_key: str,
 24        secret_key: str,
 25        base_url: str,
 26        version: str,
 27        timeout: int,
 28        session: httpx.Client,
 29    ):
 30        self._public_key = public_key
 31        self._secret_key = secret_key
 32        self._base_url = base_url
 33        self._version = version
 34        self._timeout = timeout
 35        self._session = session
 36
 37    def generate_headers(self):
 38        return {
 39            "Authorization": "Basic "
 40            + b64encode(
 41                f"{self._public_key}:{self._secret_key}".encode("utf-8")
 42            ).decode("ascii"),
 43            "Content-Type": "application/json",
 44            "x_langfuse_sdk_name": "python",
 45            "x_langfuse_sdk_version": self._version,
 46            "x_langfuse_public_key": self._public_key,
 47        }
 48
 49    def batch_post(self, **kwargs) -> httpx.Response:
 50        """Post the `kwargs` to the batch API endpoint for events"""
 51        log = logging.getLogger("langfuse")
 52        log.debug("uploading data: %s", kwargs)
 53
 54        res = self.post(**kwargs)
 55        return self._process_response(
 56            res, success_message="data uploaded successfully", return_json=False
 57        )
 58
 59    def post(self, **kwargs) -> httpx.Response:
 60        """Post the `kwargs` to the API"""
 61        log = logging.getLogger("langfuse")
 62        url = self._remove_trailing_slash(self._base_url) + "/api/public/ingestion"
 63        data = json.dumps(kwargs, cls=EventSerializer)
 64        log.debug("making request: %s to %s", data, url)
 65        headers = self.generate_headers()
 66        res = self._session.post(
 67            url, content=data, headers=headers, timeout=self._timeout
 68        )
 69
 70        if res.status_code == 200:
 71            log.debug("data uploaded successfully")
 72
 73        return res
 74
 75    def _remove_trailing_slash(self, url: str) -> str:
 76        """Removes the trailing slash from a URL"""
 77        if url.endswith("/"):
 78            return url[:-1]
 79        return url
 80
 81    def _process_response(
 82        self, res: httpx.Response, success_message: str, *, return_json: bool = True
 83    ) -> Union[httpx.Response, Any]:
 84        log = logging.getLogger("langfuse")
 85        log.debug("received response: %s", res.text)
 86        if res.status_code in (200, 201):
 87            log.debug(success_message)
 88            if return_json:
 89                try:
 90                    return res.json()
 91                except json.JSONDecodeError:
 92                    raise APIError(res.status_code, "Invalid JSON response received")
 93            else:
 94                return res
 95        elif res.status_code == 207:
 96            try:
 97                payload = res.json()
 98                errors = payload.get("errors", [])
 99                if errors:
100                    raise APIErrors(
101                        [
102                            APIError(
103                                error.get("status"),
104                                error.get("message", "No message provided"),
105                                error.get("error", "No error details provided"),
106                            )
107                            for error in errors
108                        ]
109                    )
110                else:
111                    return res.json() if return_json else res
112            except json.JSONDecodeError:
113                raise APIError(res.status_code, "Invalid JSON response received")
114
115        try:
116            payload = res.json()
117            raise APIError(res.status_code, payload)
118        except (KeyError, ValueError):
119            raise APIError(res.status_code, res.text)
120
121
122class APIError(Exception):
123    def __init__(self, status: Union[int, str], message: str, details: Any = None):
124        self.message = message
125        self.status = status
126        self.details = details
127
128    def __str__(self):
129        msg = "{0} ({1}): {2}"
130        return msg.format(self.message, self.status, self.details)
131
132
133class APIErrors(Exception):
134    def __init__(self, errors: List[APIError]):
135        self.errors = errors
136
137    def __str__(self):
138        errors = ", ".join(str(error) for error in self.errors)
139
140        return f"[Langfuse] {errors}"