Skip to main content

Create a REST Dispatch Integration

This tutorial builds on what we learned in Get Sandbox Dispatches and describes how to integrate with the Voltus REST API to fetch dispatches on an ongoing basis.

You'll need an API key for this tutorial. You should have two API keys, one each for production and sandbox. Your production API key is configured to have access to specific sites. Contact your Voltus support team to add or remove sites.

To build a dispatch integration, you'll make requests to the Voltus API at a regular interval. With each request, you'll receive a list of active, upcoming, and recent dispatches for your sites. It's then up to you to take the appropriate action on your sites.

The polling interval should be determined by the programs in which your sites are registered. Some programs give as much as a day's notice, whereas other programs have much shorter notice periods. In general, you should not need to poll more frequently than once per minute.

We provide source code examples in-line. You can obtain this source code from Voltus API Examples.

Poll for dispatches​

This code shows how to continuously poll for dispatches and send updates to sites where appropriate. We use the public sandbox environment, which always returns the same single dispatch:

"""
Skeleton program which continually polls for new dispatches
"""

import os
import urllib
import datetime
import time
from typing import Dict

import requests
from dateutil import parser as date_parser


VOLTUS_API_URL = os.getenv("VOLTUS_API_URL", "https://sandbox.voltus.co")
VOLTUS_API_KEY = os.getenv("VOLTUS_API_KEY", "secret")
POLLING_INTERVAL_SECONDS = 30

s = requests.Session()
s.headers.update({"X-Voltus-API-Key": VOLTUS_API_KEY, "Accept": "application/json"})
managed_dispatches: Dict[str, Dict] = {}  # dispatch id -> dispatch info
cancelled_dispatches = set()


def create_dispatch(dispatch_info: Dict):
    # TODO: Fill in with your company-specific logic
    #       Remember that this could get called twice if the service gets restarted
    #       Remember that end_time could be None
    #       sites could have one sites, many sites, or all of your sites
    print(
        "Dispatch with ID {} starting at {} and ending at {} for sites {} created".format(
            dispatch_info["id"],
            dispatch_info["start_time"],
            dispatch_info.get("end_time", "not specified"),
            dispatch_info["sites"],
        )
    )


def update_dispatch(dispatch_info: Dict):
    # TODO: Fill in with your company-specific logic
    #       start_time and end_time can change to be earlier or later
    #       Remember that this could get called twice if the service is restarted or the request otherwise fails
    print(
        "Dispatch with ID {} starting at {} and ending at {} for sites {} updated".format(
            dispatch_info["id"],
            dispatch_info["start_time"],
            dispatch_info.get("end_time", "not specified"),
            dispatch_info["sites"],
        )
    )


def cancel_dispatch(dispatch_info: Dict):
    # TODO: Fill in with your company-specific logic to cancel this dispatch, likely by ID
    #       Remember that this could get called twice if the service is restarted
    print("Dispatch with ID {} cancelled".format(dispatch_info["id"]))


if __name__ == "__main__":
    while True:
        start_time = datetime.datetime.now(datetime.timezone.utc)

        url = urllib.parse.urljoin(VOLTUS_API_URL, "/2022-04-15/dispatches")
        resp = s.get(url)
        resp.raise_for_status()  # raise an exception if request fails
        dispatches = resp.json()
        for dispatch_info in dispatches["dispatches"]:
            # convert start/end time to proper datetime
            dispatch_info["start_time"] = date_parser.parse(dispatch_info["start_time"])
            if "end_time" in dispatch_info and dispatch_info["end_time"] is not None:
                dispatch_info["end_time"] = date_parser.parse(dispatch_info["end_time"])
            else:
                dispatch_info["end_time"] = None

            dispatch_id = dispatch_info["id"]
            now = datetime.datetime.now(datetime.timezone.utc)

            # if we've already cancelled this dispatch, skip it
            if dispatch_id in cancelled_dispatches:
                continue

            # send a cancelled message if the dispatch is over or not authorized
            # regardless of whether we know about it or not (to protect against
            # this service restarting or similar), as long as it has not been
            # previously marked as cancelled
            if (
                dispatch_info.get("end_time") and dispatch_info["end_time"] < now
            ) or not dispatch_info["authorized"]:
                cancel_dispatch(dispatch_info)
                cancelled_dispatches.add(dispatch_id)
                managed_dispatches.pop(dispatch_id, None)
            # if this dispatch hasn't been seen before, create it
            elif dispatch_id not in managed_dispatches:
                create_dispatch(dispatch_info)
                managed_dispatches[dispatch_id] = dispatch_info
            # otherwise, update it
            else:
                if managed_dispatches[dispatch_id] != dispatch_info:
                    update_dispatch(dispatch_info)
                    managed_dispatches[dispatch_id].update(dispatch_info)

        # sleep for 30 seconds minus however long it took to process the dispatches
        elapsed = datetime.datetime.now(datetime.timezone.utc) - start_time
        time.sleep(max(0, POLLING_INTERVAL_SECONDS - elapsed.total_seconds()))

[get the code].

It should produce a message recording a newly created dispatch:

Dispatch with ID asdf starting at 2023-07-31 16:38:59.739840+00:00 and ending at 2023-07-31 17:38:59.739840+00:00 for sites [{'name': 'Site 1', 'id': 'asdf', 'customer_location_id': '1', 'commitment': 12}, {'name': 'Site 2', 'id': 'lkjh', 'customer_location_id': '2', 'commitment': 40}] created

Then a stream of updates that looks like this:

Dispatch with ID asdf starting at 2023-07-31 16:39:29.467542+00:00 and ending at 2023-07-31 17:39:29.467541+00:00 for sites [{'name': 'Site 1', 'id': 'asdf', 'customer_location_id': '1', 'commitment': 12}, {'name': 'Site 2', 'id': 'lkjh', 'customer_location_id': '2', 'commitment': 40}] updated

This is because the public sandbox always returns a single dispatch, scheduled a short period in the future.

Let’s create a more realistic scenario by using our private sandbox API key to create a dispatch scenario, then watch for updates with a properly configured version of the above program.

Running a Sandbox Dispatch Simulation​

We will create a basic dispatch simulation in sandbox.

Change the VOLTUS_API_KEY to the Sandbox API Key voltus sent in code or by setting the environment variable with:

On Linux or MacOS:

export VOLTUS_API_URL = "https://sandbox.voltus.co"
export VOLTUS_API_KEY = "<SANDBOX_API_KEY>"

Or on Windows Powershell:

$Env:VOLTUS_API_URL = "https://sandbox.voltus.co"
$Env:VOLTUS_API_KEY = "<SANDBOX_API_KEY>"

Run the dispatch scenario by running the curl command with a near-future notification_time at Sandbox Dispatch Simulations - Basic Dispatch. Before running, you'll need to change the notification_time, start_time, and end_time to now or some time in the near future.

Now, if we re-run the program above, we should get something like the following as output:

Dispatch with ID r4y5 starting at 2024-12-01T16:50:00+00:00 and ending at 2024-12-01T17:00:00+00:00 for sites [{'name': 'My Site', 'id': 'd4z9', 'customer_location_id': 'democustomerlocationid', 'drop_by': 5},{'name': 'Another Site', 'id': '9jeq', 'customer_location_id': 'democustomerlocationid2', 'drop_by': 50},] created

Next, run each dispatch simulations outlined in Sandbox Dispatch Simulations and confirm your software works as expected for each scenario.

Connecting to Production​

Once you've confirmed your software works as expected with the sandbox dispatch simulations, test your connection to the production API. Set the VOLTUS_API_KEY environment variable to your production API key.

On Linux or MacOS:

export VOLTUS_API_URL = "https://api.voltus.co"
export VOLTUS_API_KEY = "<VOLTUS_API_KEY>"

Or on Windows Powershell:

$Env:VOLTUS_API_URL = "https://api.voltus.co"
$Env:VOLTUS_API_KEY = "<VOLTUS_API_KEY>"

This will make the service contact the production version of the Voltus API. Now, create a test dispatch by running the following via curl (set the start and end times to sometime in the future before doing so, otherwise the request will fail):

curl --request POST \
  --url https://api.voltus.co/2022-04-15/dispatches \
  --header 'Content-Type: application/json' \
  --header "X-Voltus-API-Key: ${VOLTUS_API_KEY}" \
  --data '{
    "start_time": "2023-07-06T19:30:00-07:00",
    "end_time": "2023-07-06T19:40:00-07:00"
}'

Now, if we re-run the program above, we should get something like the following as output (depending on your exact configuration with Voltus):

Dispatch with ID r4y5 starting at 2023-07-31 16:38:59+00:00 and ending at 2023-07-31 17:38:59+00:00 for sites [{'name': 'Test Site 1', 'id': 'd4z9', 'customer_location_id': '1', 'drop_by': 5}] created

Next steps​

Congratulations! You have successfully set up and tested a basic dispatch service.

After passing all the simulated dispatches in sandbox and confirming your connection to production with a self-scheduled test dispatch, your integration is ready to accept new sites. Contact your Sales Director and api-support@voltus.co for next steps. Each new site must still pass a dispatch verification test. Dispatch verifications are site-specific end-to-end tests in the production environment.