Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/sdk-python. Official Python SDK for openstatus
Something went wrong. Try again.
8.4 kB · 262 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263from __future__ import annotations
import osimport sysfrom types import TracebackType
import httpx
if sys.version_info >= (3, 11): from typing import Selfelse: from typing_extensions import Self
from ._client_options import ClientOptionsfrom ._meta import DEFAULT_BASE_URL, ENV_API_KEY, ENV_BASE_URL, user_agentfrom ._transport import _AsyncJsonTransport, _JsonTransportfrom .services import ( AsyncHealthServiceClient, AsyncMaintenanceServiceClient, AsyncMonitorServiceClient, AsyncNotificationServiceClient, AsyncStatusPageServiceClient, AsyncStatusReportServiceClient, HealthServiceClient, MaintenanceServiceClient, MonitorServiceClient, NotificationServiceClient, StatusPageServiceClient, StatusReportServiceClient,)
__all__ = ["AsyncOpenstatusClient", "OpenstatusClient"]
def _resolve_base_url(options: ClientOptions) -> str: if options.base_url is not None: return options.base_url env = os.environ.get(ENV_BASE_URL) return env if env else DEFAULT_BASE_URL
def _resolve_default_headers(options: ClientOptions) -> dict[str, str]: headers: dict[str, str] = { "Content-Type": "application/json", "Accept": "application/json", "User-Agent": user_agent(), } api_key = options.api_key if options.api_key is not None else os.environ.get(ENV_API_KEY) if api_key: headers["x-openstatus-key"] = api_key headers.update(options.default_headers) return headers
# ---- sync namespace holders --------------------------------------------- #
class _HealthV1Namespace: def __init__(self, transport: _JsonTransport) -> None: self.HealthService = HealthServiceClient(transport)
class _HealthNamespace: def __init__(self, transport: _JsonTransport) -> None: self.v1 = _HealthV1Namespace(transport)
class _MonitorV1Namespace: def __init__(self, transport: _JsonTransport) -> None: self.MonitorService = MonitorServiceClient(transport)
class _MonitorNamespace: def __init__(self, transport: _JsonTransport) -> None: self.v1 = _MonitorV1Namespace(transport)
class _StatusReportV1Namespace: def __init__(self, transport: _JsonTransport) -> None: self.StatusReportService = StatusReportServiceClient(transport)
class _StatusReportNamespace: def __init__(self, transport: _JsonTransport) -> None: self.v1 = _StatusReportV1Namespace(transport)
class _StatusPageV1Namespace: def __init__(self, transport: _JsonTransport) -> None: self.StatusPageService = StatusPageServiceClient(transport)
class _StatusPageNamespace: def __init__(self, transport: _JsonTransport) -> None: self.v1 = _StatusPageV1Namespace(transport)
class _MaintenanceV1Namespace: def __init__(self, transport: _JsonTransport) -> None: self.MaintenanceService = MaintenanceServiceClient(transport)
class _MaintenanceNamespace: def __init__(self, transport: _JsonTransport) -> None: self.v1 = _MaintenanceV1Namespace(transport)
class _NotificationV1Namespace: def __init__(self, transport: _JsonTransport) -> None: self.NotificationService = NotificationServiceClient(transport)
class _NotificationNamespace: def __init__(self, transport: _JsonTransport) -> None: self.v1 = _NotificationV1Namespace(transport)
# ---- async namespace holders -------------------------------------------- #
class _AsyncHealthV1Namespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.HealthService = AsyncHealthServiceClient(transport)
class _AsyncHealthNamespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.v1 = _AsyncHealthV1Namespace(transport)
class _AsyncMonitorV1Namespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.MonitorService = AsyncMonitorServiceClient(transport)
class _AsyncMonitorNamespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.v1 = _AsyncMonitorV1Namespace(transport)
class _AsyncStatusReportV1Namespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.StatusReportService = AsyncStatusReportServiceClient(transport)
class _AsyncStatusReportNamespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.v1 = _AsyncStatusReportV1Namespace(transport)
class _AsyncStatusPageV1Namespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.StatusPageService = AsyncStatusPageServiceClient(transport)
class _AsyncStatusPageNamespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.v1 = _AsyncStatusPageV1Namespace(transport)
class _AsyncMaintenanceV1Namespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.MaintenanceService = AsyncMaintenanceServiceClient(transport)
class _AsyncMaintenanceNamespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.v1 = _AsyncMaintenanceV1Namespace(transport)
class _AsyncNotificationV1Namespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.NotificationService = AsyncNotificationServiceClient(transport)
class _AsyncNotificationNamespace: def __init__(self, transport: _AsyncJsonTransport) -> None: self.v1 = _AsyncNotificationV1Namespace(transport)
# ---- root clients ------------------------------------------------------- #
class OpenstatusClient: """Sync entry point for the Openstatus API.
The client constructs its own `httpx.Client` unless one is supplied via `ClientOptions.http_client`. When the client owns its underlying transport it will close it on `.close()` and on context-manager exit. """
def __init__(self, options: ClientOptions | None = None) -> None: opts = options if options is not None else ClientOptions() self._owns_http_client = opts.http_client is None self._http: httpx.Client = ( opts.http_client if opts.http_client is not None else httpx.Client(timeout=opts.timeout) ) transport = _JsonTransport( self._http, _resolve_base_url(opts), _resolve_default_headers(opts), ) self.health = _HealthNamespace(transport) self.monitor = _MonitorNamespace(transport) self.status_report = _StatusReportNamespace(transport) self.status_page = _StatusPageNamespace(transport) self.maintenance = _MaintenanceNamespace(transport) self.notification = _NotificationNamespace(transport)
def close(self) -> None: if self._owns_http_client: self._http.close()
def __enter__(self) -> Self: return self
def __exit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None, ) -> None: self.close()
class AsyncOpenstatusClient: """Async entry point for the Openstatus API."""
def __init__(self, options: ClientOptions | None = None) -> None: opts = options if options is not None else ClientOptions() self._owns_http_client = opts.async_http_client is None self._http: httpx.AsyncClient = ( opts.async_http_client if opts.async_http_client is not None else httpx.AsyncClient(timeout=opts.timeout) ) transport = _AsyncJsonTransport( self._http, _resolve_base_url(opts), _resolve_default_headers(opts), ) self.health = _AsyncHealthNamespace(transport) self.monitor = _AsyncMonitorNamespace(transport) self.status_report = _AsyncStatusReportNamespace(transport) self.status_page = _AsyncStatusPageNamespace(transport) self.maintenance = _AsyncMaintenanceNamespace(transport) self.notification = _AsyncNotificationNamespace(transport)
async def aclose(self) -> None: if self._owns_http_client: await self._http.aclose()
async def __aenter__(self) -> Self: return self
async def __aexit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None, ) -> None: await self.aclose()