Rockbox open source high quality audio player as a Music Player Daemon
mpris
rockbox
mpd
libadwaita
audio
rust
zig
deno
1"""Plugin protocol — Jellyfin-style install/uninstall lifecycle."""
2
3from __future__ import annotations
4
5from collections.abc import Awaitable
6from dataclasses import dataclass
7from typing import Any, Protocol, runtime_checkable
8
9from .events import EventEmitter
10
11
12@dataclass
13class PluginContext:
14 """Handed to a plugin's :meth:`install`. Lets it issue raw GraphQL and listen for events."""
15
16 query: QueryFn
17 events: EventEmitter
18
19
20class QueryFn(Protocol):
21 async def __call__(
22 self, gql: str, variables: dict[str, Any] | None = None, /
23 ) -> Any: ...
24
25
26@runtime_checkable
27class RockboxPlugin(Protocol):
28 """Anything implementing this Protocol can be loaded with :meth:`RockboxClient.use`."""
29
30 name: str
31 version: str
32 description: str | None
33
34 def install(
35 self, context: PluginContext
36 ) -> None | Awaitable[None]: ...
37
38 def uninstall(self) -> None | Awaitable[None]: # type: ignore[empty-body]
39 ...
40
41
42class PluginRegistry:
43 """Tracks installed plugins by ``name``. Names must be unique within a client."""
44
45 def __init__(self) -> None:
46 self._plugins: dict[str, RockboxPlugin] = {}
47
48 async def register(self, plugin: RockboxPlugin, context: PluginContext) -> None:
49 if plugin.name in self._plugins:
50 raise ValueError(f"Plugin {plugin.name!r} is already installed")
51 result = plugin.install(context)
52 if hasattr(result, "__await__"):
53 await result # type: ignore[misc]
54 self._plugins[plugin.name] = plugin
55
56 async def unregister(self, name: str) -> None:
57 plugin = self._plugins.pop(name, None)
58 if plugin is None:
59 return
60 uninstall = getattr(plugin, "uninstall", None)
61 if uninstall is None:
62 return
63 result = uninstall()
64 if hasattr(result, "__await__"):
65 await result
66
67 def has(self, name: str) -> bool:
68 return name in self._plugins
69
70 def list(self) -> list[RockboxPlugin]:
71 return list(self._plugins.values())