dev #3

Merged
gitark merged 84 commits from dev into main 2026-06-12 16:00:25 +03:00
3 changed files with 42 additions and 10 deletions
Showing only changes of commit 0b92e342e5 - Show all commits

View File

@@ -1,7 +1,8 @@
from typing import Optional
from requests import Session
from requests import HTTPError, Session
from oxi.adapter import OxiAdapter
from oxi.exception import OxiAPIError
from .node import Node
@@ -42,6 +43,9 @@ class OxiAPI:
return self._session.close()
def reload(self):
reload_response = self._session.get(f"{self.base_url}/reload")
reload_response.raise_for_status()
try:
reload_response = self._session.get(f"{self.base_url}/reload")
reload_response.raise_for_status()
except HTTPError as e:
raise OxiAPIError.from_http_error(e, context="Reload Oxidized") from e
return reload_response.status_code

View File

@@ -1,4 +1,16 @@
from typing import Optional
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from requests import HTTPError
_STATUS_MESSAGES: dict[int, str] = {
401: "Unauthorized",
403: "Forbidden",
500: "Internal Server Error",
502: "Bad Gateway",
503: "Service Unavailable",
504: "Gateway Timeout",
}
class OxiAPIError(Exception):
@@ -10,3 +22,13 @@ class OxiAPIError(Exception):
if self.status_code is not None:
return f"OxiAPIError: {self.args[0]} (HTTP {self.status_code})"
return f"OxiAPIError: {self.args[0]}"
@classmethod
def from_http_error(cls, e: "HTTPError", context: str = "") -> "OxiAPIError":
status = e.response.status_code
if status == 404:
message = f"{context} not found" if context else "Not found"
else:
base = _STATUS_MESSAGES.get(status) or e.response.reason or f"HTTP {status}"
message = f"{context}: {base}" if context else base
return cls(message, status)

View File

@@ -1,5 +1,9 @@
from typing import TYPE_CHECKING
from requests import HTTPError
from oxi.exception import OxiAPIError
from .view import NodeView
@@ -7,18 +11,20 @@ if TYPE_CHECKING:
from requests import Session
# TODO: Add type hints
class Node:
def __init__(self, session: "Session", base_url: str):
self._session = session
self._base_url = base_url
def __call__(self, name: str) -> NodeView:
url = f"{self._base_url}/node/show/{name}"
if not url.endswith(".json"):
url += ".json"
response = self._session.get(url)
response.raise_for_status()
try:
url = f"{self._base_url}/node/show/{name}"
if not url.endswith(".json"):
url += ".json"
response = self._session.get(url)
response.raise_for_status()
except HTTPError as e:
raise OxiAPIError.from_http_error(e, context=f"Node {name}") from e
return NodeView(
session=self._session, base_url=self._base_url, data=response.json()
)