LogalDeveloper/Owlbot
0
Fork 0
Code Issues 13 Actions Packages Releases 3 Wiki Activity

Updated Owncast clients to v0.2.5 and added spec-coverage and integration tests to enforce parity with the OpenAPI spec.
CI / Formatting (push) Successful in 6s
Details
CI / Linting (push) Successful in 5s
Details
CI / Tests (Python 3.12) (push) Successful in 2m55s
Details
CI / Tests (Python 3.13) (push) Successful in 2m46s
Details
CI / Tests (Python 3.14) (push) Successful in 2m42s
Details
CI / Type Checking (push) Successful in 10s
Details
CI / Spelling (push) Successful in 9s
Details

Browse Source
This commit is contained in:
LogalDeveloper
2026-04-24 16:57:11 -04:00
parent 44069793f8
commit f20faa317e
17 changed files with 7826 additions and 808 deletions
Download Patch File Download Diff File Expand all files Collapse all files
owlbot/owncast_http.py
+93 -30
View File
@@ -63,26 +63,108 @@ def _extract_error(text: str) -> str:
return text
async def post_with_envelope(
async def request_with_envelope(
http: HttpClient,
base_url: str,
endpoint: str,
data: dict[str, Any] | None = None,
data: dict[str, Any] | aiohttp.FormData | None = None,
*,
logger: logging.Logger,
auth: aiohttp.BasicAuth | None = None,
headers: dict[str, str] | None = None,
method: str = "POST",
) -> str:
"""Send a body-carrying request and unwrap the Owncast envelope.
The Owncast envelope is the ``{"success", "message"}`` JSON object
returned by most admin and integration endpoints.
:param data: Request body. A ``dict`` is sent as JSON; an
:class:`aiohttp.FormData` is sent as ``multipart/form-data`` with
aiohttp picking the boundary. ``None`` sends no body.
:param method: HTTP method to use (``POST`` by default; also ``PUT``,
``DELETE``, ``PATCH``).
:raises OwncastError: If the request fails or the server reports failure.
"""
url = f"{base_url}{endpoint}"
logger.debug("%s %s", method, endpoint)
request_kwargs: dict[str, Any] = (
{"data": data} if isinstance(data, aiohttp.FormData) else {"json": data}
)
try:
async with http.session.request(
method,
url,
headers=headers,
auth=auth,
allow_redirects=False,
**request_kwargs,
) as response:
if response.status >= HTTPStatus.BAD_REQUEST:
text = await response.text()
message = _extract_error(text)
logger.error(
"Error %d on %s %s: %s",
response.status,
method,
endpoint,
message,
)
raise OwncastError(response.status, message)
logger.debug("%s %s -> %d", method, 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 %s %s: %s", method, endpoint, text
)
raise OwncastError(response.status, text) from None
if isinstance(result, dict):
if "error" in result:
logger.error(
"Error on %s %s: %s", method, 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 %s %s: %s", method, endpoint, message)
raise OwncastError(response.status, message)
if "success" in result:
return str(result.get("message", ""))
logger.warning(
"Unknown response on %s %s: %s", method, endpoint, result
)
return ""
return ""
except aiohttp.ClientError as e:
logger.exception("Connection error on %s %s.", method, endpoint)
raise OwncastError(0, str(e)) from e
async def get_text(
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,
) -> str:
"""Send a POST and unwrap the Owncast ``{"success", "message"}`` envelope.
"""Send a GET and return the raw response body as text.
:raises OwncastError: If the request fails or the server reports failure.
Used for endpoints that return non-JSON payloads (e.g. Prometheus metrics).
:raises OwncastError: If the request fails.
"""
url = f"{base_url}{endpoint}"
logger.debug("POST %s", endpoint)
logger.debug("GET %s", endpoint)
try:
async with http.session.post(
async with http.session.get(
url,
json=data,
params=params,
headers=headers,
auth=auth,
allow_redirects=False,
@@ -91,32 +173,13 @@ async def post_with_envelope(
text = await response.text()
message = _extract_error(text)
logger.error(
"Error %d on POST %s: %s", response.status, endpoint, message
"Error %d on GET %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 ""
logger.debug("GET %s -> %d", endpoint, response.status)
return await response.text()
except aiohttp.ClientError as e:
logger.exception("Connection error on POST %s.", endpoint)
logger.exception("Connection error on GET %s.", endpoint)
raise OwncastError(0, str(e)) from e