Base

@classmethod
def __subclasshook__(cls, C: type) -> bool:
    """
    Determine if a class is considered a subclass of BaseEndpoint.

    This method is used to implement a custom subclass check. A class
    is considered a subclass of BaseEndpoint if it has an 'invoke' method.

    Args:
        C: The class to check.

    Returns:
        bool or NotImplemented: True if the class is a subclass, False if it isn't,
                                or NotImplemented if the check is inconclusive.
    """
    if cls is Endpoint:
        if any("invoke" in B.__dict__ for B in C.__mro__):
            return True
    return NotImplemented

@staticmethod
def create_payload(*args: Any, **kwargs: Any) -> Any:
    """
    Create a payload for the endpoint invocation.

    This static method should be implemented by subclasses to define
    how the payload is created based on the given arguments. Ideally,
    subclasses should conform to the conventions of existing endpoint types
    (for example taking a `user_message: str | list[ContentItem]` param),
    but this is not strictly enforced at the typing level.

    Args:
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments.

    Returns:
        NotImplemented: This method returns NotImplemented in the base class.
    """
    return NotImplemented

@abstractmethod
def invoke(self, payload: dict) -> InvocationResponse:
    """Call a model and return a full parsed response with error handling

    !!! info
        We strongly encourage to use the
        [`llmeter_invoke`](llmeter.endpoints.base.Endpoint.llmeter_invoke) decorator to implement
        your invoke method with proper orchestration and error handling.

    `Endpoint.invoke` should:

    1. Call `prepare_payload` to transform the input payload
    2. Invoke your actual target endpoint
    3. Parse the results onto an
        [`InvocationResponse`](llmeter.endpoints.base.InvocationResponse) object (preferably
        via [`process_raw_response`](llmeter.endpoints.base.Endpoint.process_raw_response))
    4. Populate `.error` and as many other response fields as possible, in the event that an
        error occurs during model calling or response processing

    The `llmeter_invoke` decorator handles this flow for you - so you'll need to re-implement
    the steps if you choose not to use it.

    Args:
        payload: The input payload for the model.

    Returns:
        response: The final `InvocationResponse`, including all the information that could be
            parsed from the API response - even in case of an error (when the ``error`` field
            should also be set)
    """
    raise NotImplementedError

@classmethod
def llmeter_invoke(
    cls,
    call_endpoint: Callable[..., TRawResponse],
) -> Callable[..., InvocationResponse]:
    """Wrap a raw model API call with pre+postprocessing and error handling

    This decorator wraps around a function that *only* does the core model call, to add the
    full range of steps that LLMeter Endpoints are expected to handle as part of `invoke`:

    1. **Before** starting the response timer, calls your class'
        [`prepare_payload`](llmeter.endpoints.base.Endpoint.prepare_payload) method to
        transform the input payload, if required
    2. Initialises an [`InvocationResponse`](llmeter.endpoints.base.InvocationResponse) with
        the timestamp of the request.
    3. Calls the wrapped function to fetch the raw API response
    4. Calls your class'
        [`process_raw_response`](llmeter.endpoints.base.Endpoint.process_raw_response) method
        to incrementally parse fields from the raw response to the target `InvocationResponse`
    5. In case of any unhandled errors during API call or response processing, logs and sets
        `response.error`
    6. Automatically backfills the following fields on the parsed response, if missing:
        - `id` (as a generated UUID)
        - `input_payload` (the final payload sent to the API)
        - `input_prompt` (via
            [`_parse_payload`](llmeter.endpoints.base.Endpoint._parse_payload) method)
        - `time_to_last_token`

    Args:
        call_endpoint: The function to wrap. Should be a method that takes a `payload: dict`
            and returns a `raw_response` object for input to `process_raw_response`

    Returns:
        A wrapped function that implements the full `invoke` logic.
    """

    @functools.wraps(call_endpoint)
    def wrapper(self: "Endpoint", payload: dict) -> InvocationResponse:
        prepared = self.prepare_payload(payload)
        # Snapshot before the API call for _parse_payload, which runs after
        # the inner invoke — by which point the client may have mutated the dict.
        saved_payload = copy.deepcopy(prepared)
        default_response_id = uuid4().hex
        response = InvocationResponse(
            id=default_response_id,
            request_time=datetime.now(timezone.utc),
            response_text=None,
        )
        start_t = time.perf_counter()
        try:
            raw_response: TRawResponse = call_endpoint(self, prepared)
            self.process_raw_response(raw_response, start_t, response)
            default_end_t = time.perf_counter()
        except Exception as e:
            default_end_t = time.perf_counter()
            logger.exception("Endpoint invocation failed: %s", response.error or e)
            if not response.error:
                response.error = str(e)

        if response.id is None:
            # Just in case user's parsing logic accidentally cleared the default ID provided:
            response.id = default_response_id

        if response.time_to_last_token is None and response.error is None:
            response.time_to_last_token = default_end_t - start_t

        if response.input_payload is None:
            response.input_payload = prepared
        if response.input_prompt is None:
            try:
                response.input_prompt = self._parse_payload(saved_payload)
            except Exception:
                logger.debug("_parse_payload failed; leaving input_prompt as None")

        return response

    # Add a private marker to indicate that the wrapping happened:
    # (We don't currently use this for anything except unit tests)
    wrapper._is_llmeter_invoke = True  # type: ignore
    return wrapper

@classmethod
def load(cls, endpoint_config: dict) -> "Endpoint":  # type: ignore
    """Load an endpoint configuration from a dictionary.

    This class method reads a dictionary containing an endpoint configuration,
    determines the appropriate endpoint class, and instantiates it with the
    loaded configuration.

    !!! warning "Deprecated"
        This supports the legacy `{"endpoint_type": ...}` format. New code should
        use [`load_object`][llmeter.serialization.load_object] with dicts produced by
        [`dump_object`][llmeter.serialization.dump_object].

    Args:
        endpoint_config (dict): A dictionary containing the endpoint configuration.
            Must include at minimum an `endpoint_type` key.

    Returns:
        Endpoint: An instance of the appropriate endpoint class, initialized
                  with the configuration from the dictionary.
    """
    endpoint_type = endpoint_config.pop("endpoint_type")
    endpoint_module = importlib.import_module("llmeter.endpoints")
    endpoint_class = getattr(endpoint_module, endpoint_type)
    return endpoint_class(
        **_filter_legacy_ctor_kwargs(endpoint_class, endpoint_config)
    )

@classmethod
def load_from_file(cls, path: ReadablePathLike) -> "Endpoint":
    """Load an endpoint configuration from a JSON file.

    This class method reads a JSON file containing an endpoint configuration,
    determines the appropriate endpoint class, and instantiates it with the
    loaded configuration.

    Args:
        path (str | UPath): The path to the JSON configuration file.

    Returns:
        Endpoint: An instance of the appropriate endpoint class, initialized
                  with the configuration from the file.
    """
    path = ensure_path(path)
    with path.open("r") as f:
        data = json.load(f)
    if "__llmeter_class__" in data:
        return load_object(data)
    endpoint_type = data.pop("endpoint_type")
    endpoint_module = importlib.import_module("llmeter.endpoints")
    endpoint_class = getattr(endpoint_module, endpoint_type)
    return endpoint_class(**_filter_legacy_ctor_kwargs(endpoint_class, data))

def prepare_payload(self, payload: dict) -> dict:
    """Transform the payload before sending it to the API.

    You can use it to enforce any transformations you need between the input dataset/payload
    and what actually gets sent to the model, that should not be counted in the response time
    measurement. For example: Setting fixed parameters required by the endpoint e.g.
    `streaming: False`.

    This method is called by the
    [`llmeter_invoke`](llmeter.endpoints.base.Endpoint.llmeter_invoke) wrapper *before*
    starting the timer that measures response latency.

    !!! warning
        If you made a custom :meth:`invoke` implementation **without** using the
        :meth:`llmeter_invoke` decorator - check whether your implementation actually calls
        this `prepare_payload` method or not!

    The default implementation returns ``payload`` unchanged

    Args:
        payload: The raw input payload from the caller.

    Returns:
        dict: The final payload to send to the API.
    """
    return payload

@abstractmethod
def process_raw_response(
    self,
    raw_response: TRawResponse,
    start_t: float,
    response: InvocationResponse,
) -> None:
    """Parse a raw API response onto `InvocationResponse` fields

    Subclasses implement this to extract LLMeter data points (such as time to first and last
    token, output text, number of input/output tokens, etc.) from raw model responses.

    !!! warning
        If you made a custom :meth:`invoke` implementation **without** using the
        :meth:`llmeter_invoke` decorator - check whether your implementation actually calls
        this `process_raw_response` method or not!

    This function does not return a value, but is instead expected to incrementally populate
    properties on the provided draft ``response`` object.

    In this way, partial data will be stored even if an error occurs later during processing.
    For example if a stream times out, or a guardrail intervenes - we might still be able to
    capture a unique ID initially pulled from the response header.

    See [`llmeter_invoke`](llmeter.endpoints.base.Endpoint.llmeter_invoke) for more details
    about which fields of `InvocationResponse` are automatically populated for you.

    Args:
        raw_response: The raw response object (returned by your `invoke` method before it's
            wrapped with `llmeter_invoke`)
        start_t: `time.perf_counter` timestamp captured immediately before the API call.
            Use this to calculate and populate `response.time_to_last_token` and (if in
            streaming mode) `response.time_to_first_token`.
        response: The LLMeter response object to be populated **in-place**.

    Raises:
        Exception: If something goes wrong during response streaming or parsing,
            implementations can just raise an error. The :meth:`llmeter_invoke` wrapper will
            populate ``response.error`` and ``response.time_to_last_token`` if they're not set
            already.
    """
    raise NotImplementedError

def save(self, output_path: WritablePathLike) -> Path:
    """Save the endpoint configuration to a JSON file.

    .. deprecated::
        Use :meth:`~llmeter.serialization.Serializable.save_to_file` instead, which
        provides the same behavior with a consistent name across all serializable
        LLMeter objects. This alias will be removed in a future major version.

    Args:
        output_path (str | UPath): The path where the configuration file will be saved.

    Returns:
        Path: The path the file was written to.
    """
    warnings.warn(
        "Endpoint.save() is deprecated and will be removed in a future version; "
        "use Endpoint.save_to_file() instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.save_to_file(output_path)

def to_dict(self) -> dict:
    """
    Convert the endpoint configuration to a dictionary.

    Returns:
        Dict: A dictionary representation of the endpoint configuration.
    """
    endpoint_conf = {k: v for k, v in vars(self).items() if not k.startswith("_")}
    endpoint_conf["endpoint_type"] = self.__class__.__name__
    return endpoint_conf

@classmethod
def from_json(cls, json_str: str) -> "InvocationResponse":
    """Deserialize a JSON string into an `InvocationResponse`.

    This is the inverse of [`to_json`][llmeter.endpoints.base.InvocationResponse.to_json]. It
    correctly restores types that the default JSON round-trip would leave as strings or marker
    objects:

    * `datetime`-annotated fields are parsed from ISO-8601 strings back to Python `datetime`
    * `bytes`-typed fields and `__llmeter_bytes__` markers in nested payloads are restored

    For legacy compatability, any top-level keys that are *not* recognized fields are currently
    collected into [`annotations`][llmeter.endpoints.base.InvocationResponse] rather than being
    dropped or raising an error. This supports older files where CostModel callbacks wrote
    extra fields (such as `cost_*`) directly onto the response - but may be dropped in a future
    version.

    Args:
        json_str: A JSON string representation of an InvocationResponse (produced by `to_json`
            or similar).

    Returns:
        InvocationResponse: The deserialized response.

    Example:
        A round-trip can be run as follows:
        ```python
        original = InvocationResponse(response_text="hi", ...)
        restored = InvocationResponse.from_json(original.to_json())
        ```
    """
    data = json.loads(json_str, object_hook=bytes_decoder)
    restore_dataclass_types(cls, data)
    # Route any unrecognized top-level keys into `annotations` rather than failing. This keeps
    # forward/backward compatibility: e.g. older files where callbacks wrote extra fields (like
    # `cost_*`) directly onto the response, or fields written by a different LLMeter version.
    known = {f.name for f in fields(cls)}
    extras = {k: data.pop(k) for k in list(data) if k not in known}
    if extras:
        data["annotations"] = {**extras, **(data.get("annotations") or {})}
        logger.debug(
            "Loaded %d unrecognized InvocationResponse field(s) into `annotations`: %s",
            len(extras),
            ", ".join(sorted(extras)),
        )
    return cls(**data)

def to_dict(self) -> dict:
    """Return a dictionary representation of this response.

    Returns a plain `dict` produced by `dataclasses.asdict`, preserving native Python types
    (e.g. `datetime` for `request_time`).  This is suitable for programmatic access — for
    example [`RunningStats`][llmeter.utils.RunningStats] consumes this output and relies on
    `datetime` comparisons and arithmetic.

    For JSON output, use [`to_json`][llmeter.endpoints.base.InvocationResponse.to_json], (which
    delegates to [`json_default`][llmeter.serialization.json_default]
    by default, for non-JSON-serializable data types).

    Returns:
        dict: A dictionary of response fields with native Python types.
    """
    return asdict(self)

def to_json(
    self, default: Callable[[Any], Any] | None = json_default, **kwargs: Any
) -> str:
    """Serialize this response to a JSON string.

    Uses [`json_default`][llmeter.serialization.json_default] by
    default, which handles `bytes`, `datetime`, `PathLike`, and other common non-serializable
    types.

    Args:
        default: Fallback serializer passed to `json.dumps`.
        **kwargs: Additional arguments passed to `json.dumps` (e.g., `indent`, `sort_keys`).

    Returns:
        str: JSON representation of the response.
    """
    return json.dumps(asdict(self), default=default, **kwargs)