File: ingress.py

package info (click to toggle)
python-aiohasupervisor 0.4.3-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 948 kB
  • sloc: python: 4,843; sh: 37; makefile: 3
file content (33 lines) | stat: -rw-r--r-- 1,409 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
"""Ingress client for supervisor."""

from .client import _SupervisorComponentClient
from .const import ResponseType
from .models.ingress import CreateSessionOptions, IngressPanel, IngressPanels, Session


class IngressClient(_SupervisorComponentClient):
    """Handles ingress access in Supervisor.

    This includes only the APIs with fixed paths and models from Supervisor's Ingress
    API. The wildcard proxy endpoints that allow the UI to talk to addons through Core
    and Supervisor are intentionally omitted as they can't be modeled.
    """

    async def panels(self) -> dict[str, IngressPanel]:
        """Get ingress panels, returns a map of addon slug to panel info."""
        result = await self._client.get("ingress/panels")
        return IngressPanels.from_dict(result.data).panels

    async def create_session(self, options: CreateSessionOptions | None = None) -> str:
        """Create a new ingress session."""
        result = await self._client.post(
            "ingress/session",
            json=options.to_dict() if options else None,
            response_type=ResponseType.JSON,
        )
        return Session.from_dict(result.data).session

    async def validate_session(self, session: str) -> None:
        """Validate an existing ingress session."""
        body = Session(session=session).to_dict()
        await self._client.post("ingress/validate_session", json=body)