Event Polling

curl -i -X GET \
  'https://sandbox-api.shipwell.com/events?event_name=shipment.created&limit=20' \
  -H 'Authorization: YOUR_AUTHORIZATION_HEADER'

# Ensure that requests and tenacity libraries are installed (i.e. pip install requests, pip install tenacity, etc.)
import json
import pprint
import requests
from datetime import datetime, timezone, timedelta
from requests.exceptions import HTTPError, Timeout, RequestException
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type, RetryError
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse


EVENT_LIMIT_OR_PAGE_SIZE = 2  # return a max of N items per page
EVENT_NAME = "shipment.created"  # event name or type
VERBOSE = True  # verbose logging

base_path = ""
host = "sandbox-api.shipwell.com"
target_url = "https://" + host + base_path + "/events"

header_params = {
    "Accept-Type": "application/json",
    "Authorization": "YOUR_AUTHORIZATION_HEADER"
}


# Helper functions

# Retries 5 times with exponential backoff if an HTTP error or timeout occurs
@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(
        multiplier=1, min=2, max=10
    ),  # Exponential backoff (2s, 4s, 8s...)
    retry=retry_if_exception_type((HTTPError, Timeout, RequestException)),
)
def get_request_with_retries(url, headers, params=None):
    try:
        response = requests.get(
            url, headers=headers, params=params, timeout=30
        )  # Client timeout set to 30 seconds
        response.raise_for_status()  # Raise HTTPError for bad responses (4xx, 5xx)
        return response.json()  # Get the JSON response
    except HTTPError as http_err:
        print(f"HTTP error occurred: {http_err}")
        raise
    except Timeout as timeout_err:
        print(f"Timeout error: {timeout_err}")
        raise
    except RequestException as req_err:
        print(f"Request failed: {req_err}")
        raise


# Customizing retry message for the last attempt
def get_request_with_final_retry(url, headers, params=None):
    try:
        return get_request_with_retries(url, headers, params)
    except RetryError as e:
        if e.last_attempt.attempt_number == 5:
            print("Final retry attempt failed. No more retries.")
        raise e  # Re-raise the error after printing final message


def get_current_iso_8601_time() -> str:
    # Get the current UTC time
    current_utc_time = datetime.now(timezone.utc)

    # Format as ISO 8601 string
    iso_8601_string = current_utc_time.isoformat()  # 2025-10-25T13:45:30.123456+00:00
    return iso_8601_string


def get_previous_iso_8601_time(iso_time: str, hours_before: int) -> str:
    # Replace "Z" with "+00:00" to standardize the input format
    standardized_time = iso_time.replace("Z", "+00:00")

    # Parse the input ISO 8601 string to a datetime object
    input_time = datetime.fromisoformat(standardized_time)

    # Subtract the specified number of hours
    past_time = input_time - timedelta(hours=abs(hours_before))

    # Return as ISO 8601 string
    return past_time.isoformat()


def get_midnight(iso_time: str) -> str:
    # If the input iso_time is "2025-10-25T13:45:30+00:00"
    # then this function will return "2024-10-25T00:00:00+00:00"
    # aka midnight of the date/time before the date/time

    # Replace "Z" with "+00:00" to standardize the input format
    standardized_time = iso_time.replace("Z", "+00:00")

    # Parse the input ISO 8601 string to a datetime object
    input_time = datetime.fromisoformat(standardized_time)

    # Set the time to midnight (00:00:00) for the current date
    midnight = input_time.replace(hour=0, minute=0, second=0, microsecond=0)

    # Return as ISO 8601 string
    return midnight.isoformat()


def get_query_param(url, param_name):
    # Parse the URL
    parsed_url = urlparse(url)

    # Parse the query parameters
    querystring_params = parse_qs(parsed_url.query)

    # Return the parameter value (return None if the parameter is not present)
    # Since parse_qs returns a list for each parameter, we get the first element if it exists
    return querystring_params.get(param_name, [None])[0]


def add_or_replace_query_param(url, param_name, param_value):
    # Only add or replace the parameter if param_value is truthy
    if not param_value:
        return url  # Return the original URL if param_value is falsy

    # Parse the URL
    parsed_url = urlparse(url)

    # Parse the query parameters
    querystring_params = parse_qs(parsed_url.query)

    # Add or replace the specified query parameter
    querystring_params[param_name] = param_value

    # Rebuild the query string
    new_query_string = urlencode(querystring_params, doseq=True)

    # Create a new URL with the updated query string
    new_url = urlunparse(parsed_url._replace(query=new_query_string))

    return new_url


def merge_query_params(url, params):
    # Parse the URL
    parsed_url = urlparse(url)

    # Parse existing query parameters from the URL into a dictionary
    querystring_params = parse_qs(parsed_url.query)

    # Update the dictionary with new parameters,
    # replacing existing values if needed
    for key, value in params.items():
        if value:  # Only include truthy values from params
            # Set value as a list to match parse_qs format
            querystring_params[key] = [value]

    # Rebuild the query string from the updated parameters dictionary
    new_query_string = urlencode(querystring_params, doseq=True)

    # Construct a new URL with the updated query string
    new_url = urlunparse(parsed_url._replace(query=new_query_string))

    return new_url


def process_single_event(event_id, event_name, full_event_data, verbose=False) -> bool:
    # check if your database has a record of the event
    # you are currently processing the event,
    # have previously processed the event and have conditions
    # to reprocess
    # generally determine which logic to run based on the event name or type
    # if you have already, processed the event, etc.
    # (i.e. save to your own DB, update records, cancel processing,
    # skip processing, perform workflow X for event name/type Y, etc.)
    #   note: each event has a self_link property to get the full event details
    #   for a non-deleted object if needed
    #       "self_link": 'https://sandbox-api.shipwell.com/.../01J44KYQHY...'
    pass


def process_events(
    event_name, start_time, end_time, url, starting_after_event_id=None, verbose=False
) -> bool:
    processing_complete = False
    query_params = {
        "event_name": event_name,
        "limit": EVENT_LIMIT_OR_PAGE_SIZE,
        "occurred_at.gte": start_time,
        "occurred_at.lte": end_time,
    }

    # merge override existing query param values in URL
    query_target_url = merge_query_params(url=url, params=query_params)

    # ensures starting_after value is set
    # event ids are monotonically increasing by date/time
    # so this will always return the earliest id
    # in the account.
    # setting "starting_after" also ensures that the data
    # is sorted in ascending order from oldest to newest
    starting_after_event_id_value = (
        starting_after_event_id
        or get_query_param(url=query_target_url, param_name="starting_after")
        or "00000000000000000000000000"
    )
    # modify target URL to add or replace the url param
    query_target_url = add_or_replace_query_param(
        url=query_target_url,
        param_name="starting_after",
        param_value=starting_after_event_id_value,
    )

    try:
        if verbose:
            print(f"Target URL: {query_target_url}")
            print(f"Request Header: {json.dumps(header_params, indent=2)}")
            print(f"Querystring Parameters: {json.dumps(query_params, indent=2)}")

        response_wrapper = get_request_with_final_retry(
            url=query_target_url, headers=header_params, params=query_params
        )

        if verbose:
            print("Response Body: ")
            pprint.pp(response_wrapper)

        # page of events to process (already sorted by the API in
        # ascending order when starting_after used)
        event_page = response_wrapper.get("data", [])
        # process events in already sorted older to newer order
        for event in event_page:
            # process or reprocess
            process_single_event(
                event_id=event["id"],
                event_name=event["event_name"],
                full_event_data=event,
                verbose=verbose,
            )

        # Handle paginated results (i.e. N results are returned at a time
        # and if you have more than N results then you will need to
        # paginate and follow the results to process all the matching results
        links = response_wrapper.get("links", {})
        #  "links": {"next": "https://sandbox-api.shipwell.com/events/..."}
        next_page_url = links.get("next")
        if next_page_url:
            # if there are more results get the next page of events to process
            print(f"Fetching next page: {next_page_url}")
            process_events(
                event_name=event_name,
                start_time=start_time,
                end_time=end_time,
                url=next_page_url,
            )
        else:
            print(f"No additional pages of data: {query_target_url}")

        processing_complete = True
        return processing_complete
    except Exception as err:
        print(f"Request failed after retries: {err}")

    return processing_complete


try:
    # specify a date/time range within 1-48 hours
    end_time = get_current_iso_8601_time()
    # in testing you may want to increase the hours_before value below (i.e. -24 * 365)
    start_time = get_previous_iso_8601_time(iso_time=end_time, hours_before=-24)
    # last processed event id in your system
    starting_after_event_id = None

    result = process_events(
        event_name=EVENT_NAME,
        start_time=start_time,
        end_time=end_time,
        url=target_url,
        starting_after_event_id=starting_after_event_id,
        verbose=VERBOSE,
    )
    if result:
        print("Processing complete")

except Exception as err:
    print(f"Request failed after retries: {err}")