ACC-414: Support for Vehicle Events API

README.md

@@ -209,6 +209,28 @@ my_query = """
 total_network_vehicles = dimo.identity.query(query=my_query)
 ```
 
+### Vehicle Events API (DIMO Webhooks)
+
+The SDK supports calls to the Vehicle Events API, including: registering a new webhook, subscribing and unsubscribing vehicles, checking vehicles subscribed to a specific webhook, and more. To view all the available methods, check out the [Vehicle Events API Documentation here](https://docs.dimo.org/developer-platform/vehicle-events-api-webhooks).
+
+Here's a sample of how you might register a new webhook:
+
+```python
+
+new_webhook_config = {
+    "service": "Telemetry",
+    "data": "powertrainTransmissionTravelledDistance",
+    "trigger": "valueNumber > 10000",
+    "setup": "Realtime",
+    "description": "Trigger when odometer above 10000 km",
+    "target_uri": "https://my-target-uri.com/webhook",
+    "status": "Active",
+    "verification_token": "abc"
+}
+
+dimo.vehicle_events.register_webhook(developer_jwt=dev_jwt, request=new_webhook_config)
+```
+
 ## How to Contribute to the SDK
 
 You can read more about contributing [here](https://github.com/DIMO-Network/dimo-python-sdk/blob/dev-barrettk/CONTRIBUTING.md)

dimo/api/vehicle_events.py

@@ -0,0 +1,147 @@
+from dimo.errors import check_type
+
+
+class VehicleEvents:
+
+    def __init__(self, request_method, get_auth_headers):
+        self._request = request_method
+        self._get_auth_headers = get_auth_headers
+
+    def list_all_webhooks(self, developer_jwt: str):
+        """
+        Lists all webhooks for a given developer license
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        url = f"/v1/webhooks"
+        return self._request(
+            "GET", "VehicleEvents", url, headers=self._get_auth_headers(developer_jwt)
+        )
+
+    def register_webhook(self, developer_jwt: str, request: object):
+        """
+        Creates a new webhook under the developer license
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("request", request, object)
+        url = f"/v1/webhooks"
+        return self._request(
+            "POST",
+            "VehicleEvents",
+            url,
+            headers=self._get_auth_headers(developer_jwt),
+            data=request,
+        )
+
+    def webhook_signals(self, developer_jwt: str):
+        """
+        Fetches the list of signal names available for the data field
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        url = f"/v1/webhooks/signals"
+        return self._request(
+            "GET", "VehicleEvents", url, headers=self._get_auth_headers(developer_jwt)
+        )
+
+    def list_vehicle_subscriptions(self, developer_jwt: str, token_id: str):
+        """
+        Lists all webhooks that a specified vehicle token id is subscribed to
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("token_id", token_id, str)
+        url = f"/v1/webhooks/vehicles/{token_id}"
+        return self._request(
+            "GET", "VehicleEvents", url, headers=self._get_auth_headers(developer_jwt)
+        )
+
+    def list_vehicle_subscriptions_by_event(self, developer_jwt: str, webhook_id: str):
+        """
+        Lists all vehicle subscriptions for a given webhook id
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("webhook_id", webhook_id, str)
+        url = f"/v1/webhooks/{webhook_id}"
+        return self._request(
+            "GET", "VehicleEvents", url, headers=self._get_auth_headers(developer_jwt)
+        )
+
+    def update_webhook(self, developer_jwt: str, webhook_id: str, request: object):
+        """
+        Updates a webhook by a provided webhook id
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("webhook_id", webhook_id, str)
+        check_type("request", request, object)
+        url = f"/v1/webhooks/{webhook_id}"
+        return self._request(
+            "PUT",
+            "VehicleEvents",
+            url,
+            headers=self._get_auth_headers(developer_jwt),
+            data=request,
+        )
+
+    def delete_webhook(self, developer_jwt: str, webhook_id: str):
+        """
+        Deletes a webhook by a provided webhook id
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("webhook_id", webhook_id, str)
+        url = f"/v1/webhooks/{webhook_id}"
+        return self._request(
+            "DELETE",
+            "VehicleEvents",
+            url,
+            headers=self._get_auth_headers(developer_jwt),
+        )
+
+    def subscribe_all_vehicles(self, developer_jwt: str, webhook_id: str):
+        """
+        Subscribes all vehicles to a specified webhook
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("webhook_id", webhook_id, str)
+        url = f"/v1/webhooks/{webhook_id}/subscribe/all"
+        return self._request(
+            "POST", "VehicleEvents", url, headers=self._get_auth_headers(developer_jwt)
+        )
+
+    def subscribe_vehicle(self, developer_jwt: str, token_id: str, webhook_id: str):
+        """
+        Subscribes a single vehicle to a specified webhook
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("token_id", token_id, str)
+        check_type("webhook_id", webhook_id, str)
+        url = f"/v1/webhooks/{webhook_id}/subscribe/{token_id}"
+        return self._request(
+            "POST", "VehicleEvents", url, headers=self._get_auth_headers(developer_jwt)
+        )
+
+    def unsubscribe_all_vehicles(self, developer_jwt: str, webhook_id: str):
+        """
+        Unsubscribes all vehicles from a specified webhook
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("webhook_id", webhook_id, str)
+        url = f"/v1/webhooks/{webhook_id}/unsubscribe/all"
+        return self._request(
+            "DELETE",
+            "VehicleEvents",
+            url,
+            headers=self._get_auth_headers(developer_jwt),
+        )
+
+    def unsubscribe_vehicle(self, developer_jwt: str, token_id: str, webhook_id: str):
+        """
+        Unsubscribes a single vehicle from a specified webhook
+        """
+        check_type("developer_jwt", developer_jwt, str)
+        check_type("token_id", token_id, str)
+        check_type("webhook_id", webhook_id, str)
+        url = f"/v1/webhooks/{webhook_id}/unsubscribe/{token_id}"
+        return self._request(
+            "DELETE",
+            "VehicleEvents",
+            url,
+            headers=self._get_auth_headers(developer_jwt),
+        )

dimo/dimo.py

@@ -6,6 +6,7 @@
 from .api.token_exchange import TokenExchange
 from .api.trips import Trips
 from .api.valuations import Valuations
+from .api.vehicle_events import VehicleEvents
 
 from .graphql.identity import Identity
 from .graphql.telemetry import Telemetry
@@ -83,6 +84,7 @@ def __getattr__(self, name: str) -> Any:
             "valuations": (Valuations, ("request", "_get_auth_headers")),
             "identity": (Identity, ("self",)),
             "telemetry": (Telemetry, ("self",)),
+            "vehicle_events": (VehicleEvents, ("request", "_get_auth_headers")),
         }
         if name in mapping:
             cls, deps = mapping[name]

dimo/environments.py

@@ -9,6 +9,7 @@
         "Trips": "https://trips-api.dimo.zone",
         "User": "https://users-api.dimo.zone",
         "Valuations": "https://valuations-api.dimo.zone",
+        "VehicleEvents": "https://vehicle-events-api.dimo.zone",
     },
     "Dev": {
         "Attestation": "https://attestation-api.dev.dimo.zone",
@@ -20,5 +21,6 @@
         "Trips": "https://trips-api.dev.dimo.zone",
         "User": "https://users-api.dev.dimo.zone",
         "Valuations": "https://valuations-api.dev.dimo.zone",
+        "VehicleEvents": "https://vehicle-events-api.dev.dimo.zone",
     },
 }

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dimo-python-sdk"
-version = "1.3.5"
+version = "1.4.0"
 authors = [
   { name="Barrett Kowalsky", email="barrettkowalsky@gmail.com" },
 ]