navidrome/plugins/pdk/python/host/nd_host_subsonicapi.py

# Code generated by ndpgen. DO NOT EDIT.
#
# This file contains client wrappers for the SubsonicAPI host service.
# It is intended for use in Navidrome plugins built with extism-py.
#
# IMPORTANT: Due to a limitation in extism-py, you cannot import this file directly.
# The @extism.import_fn decorators are only detected when defined in the plugin's
# main __init__.py file. Copy the needed functions from this file into your plugin.

from dataclasses import dataclass
from typing import Any

import extism
import json
import base64


class HostFunctionError(Exception):
    """Raised when a host function returns an error."""
    pass


@extism.import_fn("extism:host/user", "subsonicapi_call")
def _subsonicapi_call(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


@extism.import_fn("extism:host/user", "subsonicapi_callraw")
def _subsonicapi_callraw(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


@dataclass
class SubsonicAPICallRawResult:
    """Result type for subsonicapi_call_raw."""
    content_type: str
    data: bytes


def subsonicapi_call(uri: str) -> str:
    """Call executes a Subsonic API request and returns the JSON response.

The uri parameter should be the Subsonic API path without the server prefix,
e.g., "getAlbumList2?type=random&size=10". The response is returned as raw JSON.

    Args:
        uri: str parameter.

    Returns:
        str: The result value.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "uri": uri,
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _subsonicapi_call(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])

    return response.get("responseJson", "")


def subsonicapi_call_raw(uri: str) -> SubsonicAPICallRawResult:
    """CallRaw executes a Subsonic API request and returns the raw binary response.
Designed for binary endpoints like getCoverArt and stream that return
non-JSON data. The data is base64-encoded over JSON on the wire.

    Args:
        uri: str parameter.

    Returns:
        SubsonicAPICallRawResult containing content_type, data,.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "uri": uri,
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _subsonicapi_callraw(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])

    return SubsonicAPICallRawResult(
        content_type=response.get("contentType", ""),
        data=base64.b64decode(response.get("data", "")),
    )