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", "")),
)
