Shopify - POS

import requests
from datetime import datetime, timedelta
import argparse

####### CONSTANTS (need to be filled out before using) #######
SHOPIFY_API_URL = ""
SHOPIFY_ACCESS_TOKEN = ""

VERKADA_API_URL = "https://api.verkada.com"
VERKADA_API_TOKEN = ""

CAMERA_ID = ""
CUSTOM_EVENT_UID = ""
ORGANIZATION_ID = ""

####### TO CUSTOMIZE (change as you see fit) #######
# To see possible fields: https://shopify.dev/api/admin-rest/2022-10/resources/order#resource-object
def parse_order(order: dict) -> dict:
    return {
        "item": order["line_items"][0]["title"],
        "order_number": order["name"],
        "price": float(order["total_price"]),
        "size": order["line_items"][0]["variation_name"],
    }

####### HELPERS (do not change) #######
def iso8601_to_unixms(isodate: str) -> int:
    if isodate[-1] == "Z":
        isodate = isodate[:-1]
    dt = datetime.fromisoformat(isodate)
    return int(dt.timestamp() * 1000)

def iso8601_x_minutes_ago(current_time: str, last_x_minutes: int):
    dt = datetime.fromisoformat(current_time) - timedelta(minutes=last_x_minutes)
    return dt.isoformat()

####### MAIN #######
def main(exe_date, last_x_minutes: int):

    # Get orders from Shopify API
    response = requests.get(
        f"{SHOPIFY_API_URL}/admin/api/2022-10/orders.json",
        params={
            "status": "closed",
            "updated_at_min": iso8601_x_minutes_ago(exe_date, last_x_minutes),
            "updated_at_max": exe_date
        },
        headers={"X-Shopify-Access-Token": SHOPIFY_ACCESS_TOKEN}
    )

    if response.status_code != 200:
        print(
            "Request to {} returned a {}: {}".format(SHOPIFY_API_URL, response.status_code, response.text)
        )
        return

    orders = response.json().get("orders", [])

    # Parse orders and send them to Verkada
    for order in orders:
        event = parse_order(order)
        time_ms = iso8601_to_unixms(order['processed_at'])

        create_event_response = requests.post(
            f"{VERKADA_API_URL}/cameras/v1/video_tagging/event",
            params={
                "org_id": ORGANIZATION_ID
            },
            json={
                "camera_id": CAMERA_ID,
                "time_ms": time_ms,
                "event_type_uid": CUSTOM_EVENT_UID,
                "attributes": event
            },
            headers={"x-api-key": VERKADA_API_TOKEN}
        )

        assert create_event_response.status_code == 200, create_event_response.text
    
if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Fetch data from point-of-sale services and save as custom events"
    )
    parser.add_argument(
        '-x', '--execution_date', type=str, required=False, default=datetime.now().isoformat(), help="date of execution"
    )
    parser.add_argument(
        '-l', '--last_x_minutes', type=int, required=False, default=15, help="Retrieve sales from the last X minutes"
    )

    cmdline = parser.parse_args()
    exe_date = cmdline.execution_date
    last_x_minutes = cmdline.last_x_minutes

    main(exe_date, last_x_minutes)