Refactored Owncast clients to share a private HTTP transport helper instead of inheritance.

owlbot/api/__init__.py

@@ -31,8 +31,8 @@ This package provides all the APIs modules use to interact with Owlbot:
   (re-exported from registries)
 """
 
-# Event types and dataclasses (pure data).
 # Module-scoped service wrappers (re-exported from registries).
+from owlbot.owncast_http import OwncastError
 from owlbot.registries.commands import ModuleCommands
 from owlbot.registries.events import ModuleEvents
 from owlbot.registries.routes import ModuleRoutes
@@ -97,7 +97,7 @@ from .owncast_admin_client import (
 )
 
 # Owncast client.
-from .owncast_client import OwncastClient, OwncastError
+from .owncast_client import OwncastClient
 
 # HTTP routes (module-facing).
 from .routes import (

owlbot/api/owncast_admin_client.py

@@ -22,7 +22,7 @@ from typing import TYPE_CHECKING, Any
 
 import aiohttp
 
-from .owncast_client import OwncastClient
+from owlbot.owncast_http import get_json, post_with_envelope
 
 if TYPE_CHECKING:
     from .http_client import HttpClient
@@ -173,14 +173,14 @@ class ExternalAction:
         }
 
 
-class OwncastAdminClient(OwncastClient):
+class OwncastAdminClient:
     """Async client for the Owncast Admin API.
 
     Uses HTTP Basic Authentication to access admin endpoints.
-    All methods raise :class:`~owlbot.api.owncast_client.OwncastError` on HTTP errors.
+    All methods raise :class:`~owlbot.api.OwncastError` on HTTP errors.
 
     Uses a shared :class:`~owlbot.api.http_client.HttpClient` for HTTP
-    transport (inherited from :class:`OwncastClient`).
+    transport.
     """
 
     def __init__(
@@ -193,14 +193,19 @@ class OwncastAdminClient(OwncastClient):
         :param password: Admin password.
         :param http_client: Shared HTTP client for making requests.
         """
-        super().__init__(base_url, access_token="", http_client=http_client)
+        self._base_url = base_url.rstrip("/")
+        self._http = http_client
         self._logger = logging.getLogger("owlbot.owncast_admin_client")
-        self._headers = None
         self._auth = aiohttp.BasicAuth(username, password)
         self._logger.debug(
             "Owncast admin API client initialized for: %s", self._base_url
         )
 
+    @property
+    def base_url(self) -> str:
+        """The Owncast server base URL (e.g., "https://stream.logal.dev")."""
+        return self._base_url
+
     async def get_status(self) -> dict[str, Any]:
         """Get the current server status including stream info and viewer count.
 
@@ -1003,3 +1008,31 @@ class OwncastAdminClient(OwncastClient):
         """
         self._logger.info("Setting config: %s", description)
         return await self._post(endpoint, {"value": value})
+
+    async def _post(self, endpoint: str, data: dict[str, Any] | None = None) -> str:
+        """Send a POST request to the Owncast Admin API.
+
+        :raises OwncastError: If the request fails.
+        """
+        return await post_with_envelope(
+            self._http,
+            self._base_url,
+            endpoint,
+            data,
+            logger=self._logger,
+            auth=self._auth,
+        )
+
+    async def _get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
+        """Send a GET request to the Owncast Admin API.
+
+        :raises OwncastError: If the request fails.
+        """
+        return await get_json(
+            self._http,
+            self._base_url,
+            endpoint,
+            params,
+            logger=self._logger,
+            auth=self._auth,
+        )

owlbot/api/owncast_client.py

@@ -17,54 +17,18 @@
 from __future__ import annotations
 
 import logging
-from http import HTTPStatus
 from typing import TYPE_CHECKING, Any
 
-import aiohttp
-import orjson
-from aiohttp import ContentTypeError
 from markupsafe import escape as _escape_html
 
+from owlbot.owncast_http import get_json, post_with_envelope
+
 if TYPE_CHECKING:
+    import aiohttp
+
     from .http_client import HttpClient
 
 
-def _extract_error(text: str) -> str:
-    """Extract a human-readable error from a response body.
-
-    Owncast may return ``{"error": "..."}`` or ``{"success": false, "message": "..."}``,
-    or plain text.  This helper unwraps the JSON envelope when present.
-
-    :param text: Raw response body.
-    :return: The extracted error string.
-    """
-    try:
-        data = orjson.loads(text)
-    except (ValueError, TypeError):
-        return text
-    if isinstance(data, dict):
-        if "error" in data:
-            return str(data["error"])
-        if "message" in data:
-            return str(data["message"])
-    return text
-
-
-class OwncastError(Exception):
-    """Raised when an Owncast API request fails."""
-
-    def __init__(self, status: int, message: str) -> None:
-        """Initialize the error.
-
-        :param status: HTTP status code from the failed request, or 0 if the
-            request failed due to a connection error before receiving a response.
-        :param message: Error message or response body from the server.
-        """
-        self.status = status
-        self.message = message
-        super().__init__(f"Owncast error {status}: {message}")
-
-
 class OwncastClient:
     """Async client for the Owncast Integration API.
 
@@ -231,107 +195,29 @@ class OwncastClient:
     async def _post(self, endpoint: str, data: dict[str, Any] | None = None) -> str:
         """Send a POST request to the Owncast API.
 
-        Owncast POST endpoints return ``{"success": true, "message": "..."}``.
-        This method validates the response and returns just the message string.
-
-        :param endpoint: The API endpoint path.
-        :param data: Optional JSON body to send.
-        :return: The success message string from the response.
         :raises OwncastError: If the request fails.
         """
-        url = f"{self._base_url}{endpoint}"
+        return await post_with_envelope(
-        self._logger.debug("POST %s", endpoint)
+            self._http,
-        try:
+            self._base_url,
-            async with self._http.session.post(
+            endpoint,
-                url,
+            data,
-                json=data,
+            logger=self._logger,
             headers=self._headers,
             auth=self._auth,
-                allow_redirects=False,
-            ) as response:
-                if response.status >= HTTPStatus.BAD_REQUEST:
-                    text = await response.text()
-                    message = _extract_error(text)
-                    self._logger.error(
-                        "Error %d on POST %s: %s", response.status, endpoint, message
         )
-                    raise OwncastError(response.status, message)
-                self._logger.debug("POST %s -> %d", endpoint, response.status)
-                if response.content_type == "application/json":
-                    try:
-                        result = await response.json(loads=orjson.loads)
-                    except (ValueError, ContentTypeError):
-                        text = await response.text()
-                        self._logger.exception(
-                            "Invalid JSON on POST %s: %s", endpoint, text
-                        )
-                        raise OwncastError(response.status, text) from None
-                    if isinstance(result, dict):
-                        # Is there an error field? Owncast returns
-                        # {"error": "..."} for internal errors.
-                        if "error" in result:
-                            self._logger.error(
-                                "Error on POST %s: %s", endpoint, result["error"]
-                            )
-                            raise OwncastError(response.status, result["error"])
-                        # Does the success flag indicate failure?
-                        if "success" in result and not result["success"]:
-                            message = result.get("message", "unknown error")
-                            self._logger.error(
-                                "Error on POST %s: %s", endpoint, message
-                            )
-                            raise OwncastError(response.status, message)
-                        # Is this a simple success response? Extract the message.
-                        if "success" in result:
-                            return str(result.get("message", ""))
-                    # Unknown response shape from Owncast.
-                    self._logger.warning(
-                        "Unknown response on POST %s: %s", endpoint, result
-                    )
-                    return ""
-                return ""
-        except aiohttp.ClientError as e:
-            self._logger.exception("Connection error on POST %s.", endpoint)
-            raise OwncastError(0, str(e)) from e
 
     async def _get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
         """Send a GET request to the Owncast API.
 
-        :param endpoint: The API endpoint path.
-        :param params: Optional query parameters.
-        :return: The JSON response.
         :raises OwncastError: If the request fails.
         """
-        url = f"{self._base_url}{endpoint}"
+        return await get_json(
-        self._logger.debug("GET %s", endpoint)
+            self._http,
-        try:
+            self._base_url,
-            async with self._http.session.get(
+            endpoint,
-                url,
+            params,
-                params=params,
+            logger=self._logger,
             headers=self._headers,
             auth=self._auth,
-                allow_redirects=False,
-            ) as response:
-                if response.status >= HTTPStatus.BAD_REQUEST:
-                    text = await response.text()
-                    message = _extract_error(text)
-                    self._logger.error(
-                        "Error %d on GET %s: %s", response.status, endpoint, message
         )
-                    raise OwncastError(response.status, message)
-                self._logger.debug("GET %s -> %d", endpoint, response.status)
-                try:
-                    result = await response.json(loads=orjson.loads)
-                except (ValueError, ContentTypeError):
-                    text = await response.text()
-                    self._logger.exception("Invalid JSON on GET %s: %s", endpoint, text)
-                    raise OwncastError(response.status, text) from None
-                # Is there an error field? Owncast returns
-                # {"error": "..."} for internal errors.
-                if isinstance(result, dict) and "error" in result:
-                    self._logger.error("Error on GET %s: %s", endpoint, result["error"])
-                    raise OwncastError(response.status, result["error"])
-                return result
-        except aiohttp.ClientError as e:
-            self._logger.exception("Connection error on GET %s.", endpoint)
-            raise OwncastError(0, str(e)) from e

owlbot/bot.py

@@ -24,9 +24,9 @@ from ruamel.yaml.error import YAMLError
 from . import OWNCAST_TARGET_VERSION, __version__
 from .api.config import Config
 from .api.http_client import HttpClient
-from .api.owncast_client import OwncastError
 from .http_server import HttpServer
 from .module_loader import ModuleLoader
+from .owncast_http import OwncastError
 
 if TYPE_CHECKING:
     from pathlib import Path

owlbot/owncast_http.py

@@ -0,0 +1,167 @@
+# Copyright 2026 Logan Fick
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Shared HTTP transport helpers for the Owncast API clients."""
+
+from __future__ import annotations
+
+from http import HTTPStatus
+from typing import TYPE_CHECKING, Any
+
+import aiohttp
+import orjson
+from aiohttp import ContentTypeError
+
+if TYPE_CHECKING:
+    import logging
+
+    from owlbot.api.http_client import HttpClient
+
+
+class OwncastError(Exception):
+    """Raised when an Owncast API request fails."""
+
+    def __init__(self, status: int, message: str) -> None:
+        """Initialize the error.
+
+        :param status: HTTP status code from the failed request, or 0 if the
+            request failed due to a connection error before receiving a response.
+        :param message: Error message or response body from the server.
+        """
+        self.status = status
+        self.message = message
+        super().__init__(f"Owncast error {status}: {message}")
+
+
+def _extract_error(text: str) -> str:
+    """Extract a human-readable error from an Owncast response body.
+
+    Owncast may return ``{"error": "..."}`` or
+    ``{"success": false, "message": "..."}``, or plain text.  This helper
+    unwraps the JSON envelope when present.
+    """
+    try:
+        data = orjson.loads(text)
+    except (ValueError, TypeError):
+        return text
+    if isinstance(data, dict):
+        if "error" in data:
+            return str(data["error"])
+        if "message" in data:
+            return str(data["message"])
+    return text
+
+
+async def post_with_envelope(
+    http: HttpClient,
+    base_url: str,
+    endpoint: str,
+    data: dict[str, Any] | None = None,
+    *,
+    logger: logging.Logger,
+    auth: aiohttp.BasicAuth | None = None,
+    headers: dict[str, str] | None = None,
+) -> str:
+    """Send a POST and unwrap the Owncast ``{"success", "message"}`` envelope.
+
+    :raises OwncastError: If the request fails or the server reports failure.
+    """
+    url = f"{base_url}{endpoint}"
+    logger.debug("POST %s", endpoint)
+    try:
+        async with http.session.post(
+            url,
+            json=data,
+            headers=headers,
+            auth=auth,
+            allow_redirects=False,
+        ) as response:
+            if response.status >= HTTPStatus.BAD_REQUEST:
+                text = await response.text()
+                message = _extract_error(text)
+                logger.error(
+                    "Error %d on POST %s: %s", response.status, endpoint, message
+                )
+                raise OwncastError(response.status, message)
+            logger.debug("POST %s -> %d", endpoint, response.status)
+            if response.content_type == "application/json":
+                try:
+                    result = await response.json(loads=orjson.loads)
+                except (ValueError, ContentTypeError):
+                    text = await response.text()
+                    logger.exception("Invalid JSON on POST %s: %s", endpoint, text)
+                    raise OwncastError(response.status, text) from None
+                if isinstance(result, dict):
+                    if "error" in result:
+                        logger.error("Error on POST %s: %s", endpoint, result["error"])
+                        raise OwncastError(response.status, result["error"])
+                    if "success" in result and not result["success"]:
+                        message = result.get("message", "unknown error")
+                        logger.error("Error on POST %s: %s", endpoint, message)
+                        raise OwncastError(response.status, message)
+                    if "success" in result:
+                        return str(result.get("message", ""))
+                logger.warning("Unknown response on POST %s: %s", endpoint, result)
+                return ""
+            return ""
+    except aiohttp.ClientError as e:
+        logger.exception("Connection error on POST %s.", endpoint)
+        raise OwncastError(0, str(e)) from e
+
+
+async def get_json(
+    http: HttpClient,
+    base_url: str,
+    endpoint: str,
+    params: dict[str, Any] | None = None,
+    *,
+    logger: logging.Logger,
+    auth: aiohttp.BasicAuth | None = None,
+    headers: dict[str, str] | None = None,
+) -> Any:
+    """Send a GET and return the parsed JSON body.
+
+    :raises OwncastError: If the request fails or the response cannot be parsed.
+    """
+    url = f"{base_url}{endpoint}"
+    logger.debug("GET %s", endpoint)
+    try:
+        async with http.session.get(
+            url,
+            params=params,
+            headers=headers,
+            auth=auth,
+            allow_redirects=False,
+        ) as response:
+            if response.status >= HTTPStatus.BAD_REQUEST:
+                text = await response.text()
+                message = _extract_error(text)
+                logger.error(
+                    "Error %d on GET %s: %s", response.status, endpoint, message
+                )
+                raise OwncastError(response.status, message)
+            logger.debug("GET %s -> %d", endpoint, response.status)
+            try:
+                result = await response.json(loads=orjson.loads)
+            except (ValueError, ContentTypeError):
+                text = await response.text()
+                logger.exception("Invalid JSON on GET %s: %s", endpoint, text)
+                raise OwncastError(response.status, text) from None
+            if isinstance(result, dict) and "error" in result:
+                logger.error("Error on GET %s: %s", endpoint, result["error"])
+                raise OwncastError(response.status, result["error"])
+            return result
+    except aiohttp.ClientError as e:
+        logger.exception("Connection error on GET %s.", endpoint)
+        raise OwncastError(0, str(e)) from e