feat: add dashboards endpoints. use axiom.co/api

src/axiom_py/__init__.py

@@ -28,13 +28,20 @@
     AnnotationUpdateRequest,
     AnnotationsClient,
 )
+from .dashboards import (
+    Dashboard,
+    DashboardCreateRequest,
+    DashboardUpdateRequest,
+    DashboardsClient,
+)
 from .logging import AxiomHandler
 from .structlog import AxiomProcessor
 
 # Async API
 from .client_async import AsyncClient
 from .datasets_async import AsyncDatasetsClient
 from .annotations_async import AsyncAnnotationsClient
+from .dashboards_async import AsyncDashboardsClient
 from .tokens_async import AsyncTokensClient
 from .users_async import AsyncUsersClient
 from .logging_async import AsyncAxiomHandler
@@ -57,16 +64,21 @@
     Annotation,
     AnnotationCreateRequest,
     AnnotationUpdateRequest,
+    Dashboard,
+    DashboardCreateRequest,
+    DashboardUpdateRequest,
     # Sync API
     Client,
     DatasetsClient,
     AnnotationsClient,
+    DashboardsClient,
     AxiomHandler,
     AxiomProcessor,
     # Async API
     AsyncClient,
     AsyncDatasetsClient,
     AsyncAnnotationsClient,
+    AsyncDashboardsClient,
     AsyncTokensClient,
     AsyncUsersClient,
     AsyncAxiomHandler,

src/axiom_py/client.py

@@ -22,13 +22,14 @@
     QueryKind,
 )
 from .annotations import AnnotationsClient
+from .dashboards import DashboardsClient
 from .users import UsersClient
 from .version import __version__
 from .util import from_dict, handle_json_serialization, is_personal_token
 from .tokens import TokensClient
 
 
-AXIOM_URL = "https://api.axiom.co"
+AXIOM_URL = "https://axiom.co/api"
 
 
 class PersonalTokenNotSupportedForEdgeError(Exception):
@@ -161,6 +162,7 @@ class Client:  # pylint: disable=R0903
     datasets: DatasetsClient
     users: UsersClient
     annotations: AnnotationsClient
+    dashboards: DashboardsClient
     tokens: TokensClient
     is_closed: bool = False  # track if the client has been closed (for tests)
     before_shutdown_funcs: List[Callable] = []
@@ -243,6 +245,7 @@ def __init__(
         self.datasets = DatasetsClient(self.session)
         self.users = UsersClient(self.session, is_personal_token(token))
         self.annotations = AnnotationsClient(self.session)
+        self.dashboards = DashboardsClient(self.session)
         self.tokens = TokensClient(self.session)
 
         # wrap shutdown hook in a lambda passing in self as a ref

src/axiom_py/client_async.py

@@ -21,6 +21,7 @@
 )
 from .datasets_async import AsyncDatasetsClient
 from .annotations_async import AsyncAnnotationsClient
+from .dashboards_async import AsyncDashboardsClient
 from .tokens_async import AsyncTokensClient
 from .users_async import AsyncUsersClient
 from .query import (
@@ -41,6 +42,7 @@ class AsyncClient:
     datasets: AsyncDatasetsClient
     users: AsyncUsersClient
     annotations: AsyncAnnotationsClient
+    dashboards: AsyncDashboardsClient
     tokens: AsyncTokensClient
     client: httpx.AsyncClient
 
@@ -108,6 +110,7 @@ def __init__(
         self.datasets = AsyncDatasetsClient(self.client)
         self.users = AsyncUsersClient(self.client, is_personal_token(token))
         self.annotations = AsyncAnnotationsClient(self.client)
+        self.dashboards = AsyncDashboardsClient(self.client)
         self.tokens = AsyncTokensClient(self.client)
 
     async def __aenter__(self):

src/axiom_py/dashboards.py

@@ -0,0 +1,138 @@
+"""This package provides dashboard models and methods as well as a DashboardsClient"""
+
+import ujson
+from requests import Session
+from typing import List, Optional, Dict
+from dataclasses import dataclass, asdict, field
+from .util import from_dict
+
+
+@dataclass
+class Dashboard:
+    """Represents an Axiom dashboard"""
+
+    id: str = field(init=False)
+    name: str
+    owner: str
+    refreshTime: int
+    schemaVersion: int
+    timeWindowStart: str
+    timeWindowEnd: str
+    charts: Optional[Dict] = None
+    layout: Optional[Dict] = None
+    description: Optional[str] = None
+    datasets: Optional[List[str]] = None
+    against: Optional[str] = None
+    againstTimestamp: Optional[str] = None
+    overrides: Optional[Dict] = None
+    version: Optional[str] = None
+    createdAt: Optional[str] = None
+    createdBy: Optional[str] = None
+    updatedAt: Optional[str] = None
+    updatedBy: Optional[str] = None
+
+
+@dataclass
+class DashboardCreateRequest:
+    """Request used to create a dashboard"""
+
+    name: str
+    owner: str
+    refreshTime: int
+    schemaVersion: int
+    timeWindowStart: str
+    timeWindowEnd: str
+    charts: Optional[Dict] = None
+    layout: Optional[Dict] = None
+    description: Optional[str] = None
+    datasets: Optional[List[str]] = None
+    against: Optional[str] = None
+    againstTimestamp: Optional[str] = None
+    overrides: Optional[Dict] = None
+
+
+@dataclass
+class DashboardUpdateRequest:
+    """Request used to update a dashboard"""
+
+    name: str
+    owner: str
+    refreshTime: int
+    schemaVersion: int
+    timeWindowStart: str
+    timeWindowEnd: str
+    charts: Optional[Dict] = None
+    layout: Optional[Dict] = None
+    description: Optional[str] = None
+    datasets: Optional[List[str]] = None
+    against: Optional[str] = None
+    againstTimestamp: Optional[str] = None
+    overrides: Optional[Dict] = None
+    version: Optional[str] = None
+
+
+class DashboardsClient:  # pylint: disable=R0903
+    """DashboardsClient has methods to manipulate dashboards."""
+
+    session: Session
+
+    def __init__(self, session: Session):
+        self.session = session
+
+    def get(self, id: str) -> Dashboard:
+        """
+        Get a dashboard by id.
+
+        See https://axiom.co/docs/restapi/endpoints/getDashboard
+        """
+        path = "/v1/dashboards/%s" % id
+        res = self.session.get(path)
+        decoded_response = res.json()
+        return from_dict(Dashboard, decoded_response)
+
+    def create(self, req: DashboardCreateRequest) -> Dashboard:
+        """
+        Create a dashboard with the given properties.
+
+        See https://axiom.co/docs/restapi/endpoints/createDashboard
+        """
+        path = "/v1/dashboards"
+        res = self.session.post(path, data=ujson.dumps(asdict(req)))
+        dashboard = from_dict(Dashboard, res.json())
+        return dashboard
+
+    def list(self) -> List[Dashboard]:
+        """
+        List all dashboards.
+
+        See https://axiom.co/docs/restapi/endpoints/getDashboards
+        """
+        path = "/v1/dashboards"
+        res = self.session.get(path)
+
+        dashboards = []
+        for record in res.json():
+            ds = from_dict(Dashboard, record)
+            dashboards.append(ds)
+
+        return dashboards
+
+    def update(self, id: str, req: DashboardUpdateRequest) -> Dashboard:
+        """
+        Update a dashboard with the given properties.
+
+        See https://axiom.co/docs/restapi/endpoints/updateDashboard
+        """
+        path = "/v1/dashboards/%s" % id
+        res = self.session.put(path, data=ujson.dumps(asdict(req)))
+        dashboard = from_dict(Dashboard, res.json())
+        return dashboard
+
+    def delete(self, id: str):
+        """
+        Deletes a dashboard with the given id.
+
+        See https://axiom.co/docs/restapi/endpoints/deleteDashboard
+        """
+        path = "/v1/dashboards/%s" % id
+        self.session.delete(path)

src/axiom_py/dashboards_async.py

@@ -0,0 +1,118 @@
+"""
+Async version of dashboard models and methods with AsyncDashboardsClient.
+"""
+
+import ujson
+import httpx
+from typing import List
+from dataclasses import asdict
+
+from .dashboards import (
+    Dashboard,
+    DashboardCreateRequest,
+    DashboardUpdateRequest,
+)
+from .util import from_dict
+from ._error_handling import check_response_error
+
+
+class AsyncDashboardsClient:
+    """AsyncDashboardsClient has async methods to manipulate dashboards."""
+
+    client: httpx.AsyncClient
+
+    def __init__(self, client: httpx.AsyncClient):
+        """
+        Initialize the async dashboards client.
+
+        Args:
+            client: httpx AsyncClient instance for making HTTP requests
+        """
+        self.client = client
+
+    async def get(self, id: str) -> Dashboard:
+        """
+        Asynchronously get a dashboard by id.
+
+        Args:
+            id: Dashboard identifier
+
+        Returns:
+            Dashboard object
+
+        See https://axiom.co/docs/restapi/endpoints/getDashboard
+        """
+        path = f"/v1/dashboards/{id}"
+        response = await self.client.get(path)
+        check_response_error(response.status_code, response.json())
+        return from_dict(Dashboard, response.json())
+
+    async def create(self, req: DashboardCreateRequest) -> Dashboard:
+        """
+        Asynchronously create a dashboard with the given properties.
+
+        Args:
+            req: Dashboard creation request
+
+        Returns:
+            Created Dashboard object
+
+        See https://axiom.co/docs/restapi/endpoints/createDashboard
+        """
+        path = "/v1/dashboards"
+        payload = ujson.dumps(asdict(req))
+        response = await self.client.post(path, content=payload)
+        check_response_error(response.status_code, response.json())
+        return from_dict(Dashboard, response.json())
+
+    async def list(self) -> List[Dashboard]:
+        """
+        Asynchronously list all dashboards.
+
+        Returns:
+            List of Dashboard objects
+
+        See https://axiom.co/docs/restapi/endpoints/getDashboards
+        """
+        path = "/v1/dashboards"
+        response = await self.client.get(path)
+        check_response_error(response.status_code, response.json())
+
+        dashboards = []
+        for record in response.json():
+            ds = from_dict(Dashboard, record)
+            dashboards.append(ds)
+
+        return dashboards
+
+    async def update(self, id: str, req: DashboardUpdateRequest) -> Dashboard:
+        """
+        Asynchronously update a dashboard with the given properties.
+
+        Args:
+            id: Dashboard identifier
+            req: Dashboard update request
+
+        Returns:
+            Updated Dashboard object
+
+        See https://axiom.co/docs/restapi/endpoints/updateDashboard
+        """
+        path = f"/v1/dashboards/{id}"
+        payload = ujson.dumps(asdict(req))
+        response = await self.client.put(path, content=payload)
+        check_response_error(response.status_code, response.json())
+        return from_dict(Dashboard, response.json())
+
+    async def delete(self, id: str):
+        """
+        Asynchronously delete a dashboard with the given id.
+
+        Args:
+            id: Dashboard identifier
+
+        See https://axiom.co/docs/restapi/endpoints/deleteDashboard
+        """
+        path = f"/v1/dashboards/{id}"
+        response = await self.client.delete(path)
+        check_response_error(response.status_code, response.json())