Skip to content

REST adapter (aiohttp)

rest_aiohttp

aiohttp-based implementation of RestPort.

Only the endpoints actually used by the core are exposed. The adapter optionally accepts an externally-owned aiohttp.ClientSession; when no session is provided one is created and managed internally.

AiohttpRestAdapter

Async REST client backed by aiohttp.

Parameters:

Name Type Description Default
base_url str

Home Assistant base URL (without trailing slash).

 required
token str

Long-lived access token.

 required
session ClientSession or None

Externally-owned session to reuse. When None the adapter creates its own session and closes it on close.

 None
timeout float

Request timeout in seconds.

 30.0
verify_ssl bool

Verify TLS certificates.

 True
Source code in src/haclient/infra/rest_aiohttp.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class AiohttpRestAdapter:
    """Async REST client backed by ``aiohttp``.

    Parameters
    ----------
    base_url : str
        Home Assistant base URL (without trailing slash).
    token : str
        Long-lived access token.
    session : aiohttp.ClientSession or None, optional
        Externally-owned session to reuse. When ``None`` the adapter
        creates its own session and closes it on `close`.
    timeout : float, optional
        Request timeout in seconds.
    verify_ssl : bool, optional
        Verify TLS certificates.
    """

    def __init__(
        self,
        base_url: str,
        token: str,
        *,
        session: aiohttp.ClientSession | None = None,
        timeout: float = 30.0,
        verify_ssl: bool = True,
    ) -> None:
        self._base_url = base_url.rstrip("/")
        self._token = token
        self._timeout = timeout
        self._verify_ssl = verify_ssl
        self._session = session
        self._owns_session = session is None

    @property
    def base_url(self) -> str:
        """Return the configured base URL."""
        return self._base_url

    async def _ensure_session(self) -> aiohttp.ClientSession:
        """Return the current session, creating one if necessary."""
        if self._session is None or self._session.closed:
            self._session = aiohttp.ClientSession()
            self._owns_session = True
        return self._session

    async def close(self) -> None:
        """Close the underlying HTTP session if we own it."""
        if self._owns_session and self._session is not None and not self._session.closed:
            await self._session.close()

    @property
    def _headers(self) -> dict[str, str]:
        """Return authorization and content-type headers."""
        return {
            "Authorization": f"Bearer {self._token}",
            "Content-Type": "application/json",
        }

    def _url(self, path: str) -> str:
        """Build a full URL from a relative API path."""
        if not path.startswith("/"):
            path = "/" + path
        return f"{self._base_url}{path}"

    async def _request(
        self,
        method: str,
        path: str,
        *,
        json: Any | None = None,
    ) -> Any:
        """Perform an HTTP request and return the parsed response.

        Parameters
        ----------
        method : str
            HTTP method (e.g. ``"GET"``).
        path : str
            Relative API path.
        json : Any or None, optional
            JSON-serialisable body.

        Returns
        -------
        Any
            Parsed JSON response, or raw text for non-JSON responses.

        Raises
        ------
        AuthenticationError
            On HTTP 401.
        HTTPError
            On any other HTTP error (status >= 400).
        HAClientError
            On connection failure.
        TimeoutError
            On request timeout.
        """
        session = await self._ensure_session()
        url = self._url(path)
        try:
            async with session.request(
                method,
                url,
                headers=self._headers,
                json=json,
                ssl=self._verify_ssl,
                timeout=aiohttp.ClientTimeout(total=self._timeout),
            ) as resp:
                if resp.status == 401:
                    raise AuthenticationError("Invalid or expired access token")
                if resp.status >= 400:
                    body = await resp.text()
                    raise HTTPError(resp.status, method, path, body)
                if resp.status == 200 and resp.content_type == "application/json":
                    return await resp.json()
                return await resp.text()
        except TimeoutError as err:
            raise HATimeoutError(f"Request to {path} timed out") from err
        except aiohttp.ClientError as err:
            raise HAClientError(f"HTTP request failed: {err}") from err

    async def ping(self) -> bool:
        """Verify Home Assistant is reachable.

        Returns
        -------
        bool
            ``True`` if the ``/api/`` endpoint responded successfully.
            Failures surface as exceptions from `_request`; this method
            never returns ``False``.
        """
        await self._request("GET", "/api/")
        return True

    async def get_states(self) -> list[dict[str, Any]]:
        """Return all entity states currently known to Home Assistant.

        Returns
        -------
        list of dict
            The state objects.

        Raises
        ------
        HAClientError
            If the response is not a list.
        """
        data = await self._request("GET", "/api/states")
        if not isinstance(data, list):
            raise HAClientError("Unexpected response from /api/states")
        return data

    async def get_state(self, entity_id: str) -> dict[str, Any] | None:
        """Return the state object for *entity_id*, or ``None`` if missing.

        Parameters
        ----------
        entity_id : str
            Fully-qualified entity id.

        Returns
        -------
        dict or None
            The state object, or ``None`` on HTTP 404.

        Raises
        ------
        HTTPError
            On any HTTP error other than 404.
        AuthenticationError
            On HTTP 401.
        """
        try:
            data = await self._request("GET", f"/api/states/{entity_id}")
        except HTTPError as err:
            if err.status == 404:
                return None
            raise
        if isinstance(data, dict):
            return data
        return None

    async def call_service(
        self,
        domain: str,
        service: str,
        data: dict[str, Any] | None = None,
    ) -> list[dict[str, Any]]:
        """Invoke a service via REST.

        Parameters
        ----------
        domain : str
            The service domain.
        service : str
            The service name.
        data : dict or None, optional
            Service data payload.

        Returns
        -------
        list of dict
            The list of states changed by the service call.
        """
        payload = data or {}
        result = await self._request("POST", f"/api/services/{domain}/{service}", json=payload)
        if isinstance(result, list):
            return result
        return []

base_url property 

base_url: str

Return the configured base URL.

close async 

close() -> None

Close the underlying HTTP session if we own it.

Source code in src/haclient/infra/rest_aiohttp.py 
67
68
69
70
async def close(self) -> None:
    """Close the underlying HTTP session if we own it."""
    if self._owns_session and self._session is not None and not self._session.closed:
        await self._session.close()

ping async 

ping() -> bool

Verify Home Assistant is reachable.

Returns:

Type Description
bool

True if the /api/ endpoint responded successfully. Failures surface as exceptions from _request; this method never returns False.
Source code in src/haclient/infra/rest_aiohttp.py 
144
145
146
147
148
149
150
151
152
153
154
155
async def ping(self) -> bool:
    """Verify Home Assistant is reachable.

    Returns
    -------
    bool
        ``True`` if the ``/api/`` endpoint responded successfully.
        Failures surface as exceptions from `_request`; this method
        never returns ``False``.
    """
    await self._request("GET", "/api/")
    return True

get_states async 

get_states() -> list[dict[str, Any]]

Return all entity states currently known to Home Assistant.

Returns:

Type Description
list of dict

The state objects.

Raises:

Type Description
HAClientError

If the response is not a list.
Source code in src/haclient/infra/rest_aiohttp.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def get_states(self) -> list[dict[str, Any]]:
    """Return all entity states currently known to Home Assistant.

    Returns
    -------
    list of dict
        The state objects.

    Raises
    ------
    HAClientError
        If the response is not a list.
    """
    data = await self._request("GET", "/api/states")
    if not isinstance(data, list):
        raise HAClientError("Unexpected response from /api/states")
    return data

get_state async 

get_state(entity_id: str) -> dict[str, Any] | None

Return the state object for entity_id, or None if missing.

Parameters:

Name Type Description Default
entity_id str

Fully-qualified entity id.

 required

Returns:

Type Description
dict or None

The state object, or None on HTTP 404.

Raises:

Type Description
HTTPError

On any HTTP error other than 404.
AuthenticationError

On HTTP 401.
Source code in src/haclient/infra/rest_aiohttp.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
async def get_state(self, entity_id: str) -> dict[str, Any] | None:
    """Return the state object for *entity_id*, or ``None`` if missing.

    Parameters
    ----------
    entity_id : str
        Fully-qualified entity id.

    Returns
    -------
    dict or None
        The state object, or ``None`` on HTTP 404.

    Raises
    ------
    HTTPError
        On any HTTP error other than 404.
    AuthenticationError
        On HTTP 401.
    """
    try:
        data = await self._request("GET", f"/api/states/{entity_id}")
    except HTTPError as err:
        if err.status == 404:
            return None
        raise
    if isinstance(data, dict):
        return data
    return None

call_service async 

call_service(domain: str, service: str, data: dict[str, Any] | None = None) -> list[dict[str, Any]]

Invoke a service via REST.

Parameters:

Name Type Description Default
domain str

The service domain.

 required
service str

The service name.

 required
data dict or None

Service data payload.

 None

Returns:

Type Description
list of dict

The list of states changed by the service call.
Source code in src/haclient/infra/rest_aiohttp.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
async def call_service(
    self,
    domain: str,
    service: str,
    data: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """Invoke a service via REST.

    Parameters
    ----------
    domain : str
        The service domain.
    service : str
        The service name.
    data : dict or None, optional
        Service data payload.

    Returns
    -------
    list of dict
        The list of states changed by the service call.
    """
    payload = data or {}
    result = await self._request("POST", f"/api/services/{domain}/{service}", json=payload)
    if isinstance(result, list):
        return result
    return []