Quick Python Script for processing Overland web hooks

[gloriouslyawkwardlife]

Up until now I've just had an Apache CouchDB server set up and exposed to the internet to receive the webhook packets from Overland and then would use Python to pull the data and push it into a table in MariaDB, but last night I decided to cut out the middleman altogether and build a cgi-bin application in Python to receive the hook and push it directly into the database. I wanted to build this as a Wolfram API using Wolfram Language like I've done most everything else, but Wolfram APIs only accept web form (application/x-www-form-urlencoded) and not JSON payloads.

The script drops "-1" values and replaces them with an SQL NULL, and then I made a handler for the 'motion' field, which comes across as a JSON array (a.k.a. List in Python and Wolfram).

#!/usr/bin/env python3.11

import json
import MySQLdb
import sys, os
from datetime import datetime
from collections.abc import Iterable

content_len = os.environ.get("CONTENT_LENGTH", 0)
body = sys.stdin.read(int(content_len))
input = json.loads(body)

cn = MySQLdb.connect(
    user="xxx",
    password="xxx",
    host="localhost",
    port=3306,
    database="personalinformatics",
)
cr = cn.cursor()
for l in input["locations"]:
    cr.execute(
        "REPLACE INTO overland (latitude,longitude,speed,battery_state,motion,timestamp,battery_level,speed_accuracy,horizontal_accuracy,vertical_accuracy,wifi,course,altitude,course_accuracy) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
        (
            l["geometry"]["coordinates"][1],
            l["geometry"]["coordinates"][0],
            (
                l["properties"].get("speed")
                if l["properties"].get("speed") != -1
                else None
            ),
            l["properties"].get("battery_state"),
            (
                ",".join(l["properties"].get("motion"))
                if isinstance(l["properties"].get("motion"), Iterable)
                and len(l["properties"].get("motion")) > 0
                else None
            ),
            datetime.fromisoformat(l["properties"].get("timestamp")).astimezone(),
            l["properties"].get("battery_level"),
            (
                l["properties"].get("speed_accuracy")
                if l["properties"].get("speed_accuracy") != -1
                else None
            ),
            (
                l["properties"].get("horizontal_accuracy")
                if l["properties"].get("horizontal_accuracy") != -1
                else None
            ),
            (
                l["properties"].get("vertical_accuracy")
                if l["properties"].get("vertical_accuracy") != -1
                else None
            ),
            l["properties"].get("wifi"),
            (
                l["properties"].get("course")
                if l["properties"].get("course") != -1
                else None
            ),
            l["properties"].get("altitude"),
            (
                l["properties"].get("course_accuracy")
                if l["properties"].get("course_accuracy") != -1
                else None
            ),
        ),
    )
    cn.commit()
print('Content-type: application/json\r\n\r\n{"result":"ok"}\r\n')

[gloriouslyawkwardlife]