Getting Final Mile Photos Data Into Your Database 

Curious about how to store these in a database to be accessible to the rest of our business applications? In this example we’re using and we are storing all images inside the database with foreign key constraints. You may need to alter the DDL and/or SQL to suit your specific SQL tool. 

In your production services, most modern designs would call for placing image data in what we now call “object stores”, where modern cloud services offer image and other file hosting for substantially lower costs than what you normally pay for relational database storage. Classic designs in which all data is stored in a single RDBMS using table storage partitions, user access management, and similar, are also valid, but are not covered here. In this example we show a single storage location using SQLite. 

Setting up the database and a simple “insert data” function 

To store data in a relational database we need to specify a schema for both shipment data and image data. 

import sqlite3
 
def setup_schema(conn): 
    ddl = [ 
        """ 
        CREATE TABLE IF NOT EXISTS known_shipments( 
            pro TEXT UNIQUE ON CONFLICT REPLACE, 
            status TEXT, 
            longstatus TEXT, 
            delivery_time TEXT, 
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP 
        ); 
        """, 
        """ 
        CREATE TABLE IF NOT EXISTS shipment_images( 
            known_pro TEXT REFERENCES known_shipments(pro) ON DELETE CASCADE, 
            image_identifier TEXT, 
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP, 
            image_data BLOB, 
            UNIQUE (known_pro, image_identifier) 
            ON CONFLICT REPLACE 
        ); 
        """, 
    ] 
    for d_i in ddl: 
        conn.execute(d_i) 
        conn.commit() 

def insert_data(conn, table: str, data: dict): 
    cols = list(data) 
    columns = ",".join(cols) 
    vals = ",".join(len(cols) * ["?"]) 
    query = f"INSERT INTO {table}({columns}) VALUES ({vals});" 
    conn.execute(query, [data[k] for k in cols]) 
    conn.commit() 

sq_conn = sqlite3.Connection("final_mile_test.sqlite3") 
setup_schema(sq_conn)

This code provides a function to both set up a SQLite database with proper schema, as well as a convenience function for inserting data into any named table. 

Fetch from the API then insert into your database 

From here, the function to insert shipment status into the database is straightforward: 

def get_status_to_db(conn, pro: str): 
status = get_status(pro) 
    if "errors" in status: 
        return status 
    to_insert = { 
        k: status[k] for k in ("pro", "status", "longstatus", "delivery_time") 
    } 
    insert_data(conn, "known_shipments", to_insert) 
    return status   

Because we already have a convenience function for fetching images to our local disk, shipping them to any database or object store is also straightforward: 

def get_images_to_db(conn, pro: str): 
for image_name in get_individual_images(pro, "proof"): 
        with open(image_name, "rb") as img: 
            img_data = img.read() 
            to_insert = { 
                "known_pro": pro, 
                "image_identifier": image_name, 
                "image_data": img_data, 
            } 
            insert_data(conn, "shipment_images", to_insert) 
        # remember to delete the local temporary file 
        os.unlink(image_name)

 Then to get the current shipment status and all known images for a delivery: 

def get_all_to_db(conn, pro: str): 
  status = get_status_to_db(conn, pro) 
  if "errors" not in status: 
  # can get any uploaded images any time there isn't an error 
  get_images_to_db(conn, pro)

Now to get the most updated images (if any are available), you can simply call get_all_to_db(sq_conn, PRO) to have all information we know about that shipment inserted into your local database. 

For your convenience, we have a copy of these functions with additional comments, documentation, and a simple command-line interface available on Github. 

Using Final Mile Webhooks to call your API directly 

With Final Mile Photos webhooks, instead of polling our API to determine current shipment status, you can have Liminal Network send you status updates and / or photos as they are uploaded. To configure Final Mile Photos Webhooks visit: https://account.liminalnetwork.com/account/notify-webhook 

If we re-use our code from earlier, your API can be set up to receive only the “images complete” messages status=IMAGES_COMPLETE (or error conditions can include COMPLETE_DAMAGED, COMPLETE_MISSING, NOT_AT_ADDRESS, REFUSED, or OTHER) at step 3b, which would allow you to directly use the earlier get_all_to_db(conn, ref) to fetch images at your desired speed. If you have a deferred task system, this would be a great use of it. 

For Final Mile Photos customers with many deliveries each day, if your system performs substantial processing for each image, downloading images at your convenience may reduce your web server load compared to receiving them directly from our webhooks. 

Dispatching to start, image, and end shipment handlers 

Unlike outgoing requests to Liminal Network API endpoints, incoming requests to your web server may be formed or formatted depending on your choice of libraries. We provide the below to attempt to be available to the widest variety of servers and services, as the following functions should adapt to your specific situation with little to no work. 

import base64 
import datetime 
import urllib.parse
 
def handle_request( 
    conn, request_body: Union[str, bytes], body_base64_encoded: bool 
) -> str: 
    # turn POST body into dictionary-like thing 
    if body_base64_encoded: 
        request_body = base64.b64decode( 
            request_body 
            if isinstance(request_body, bytes) 
            else request_body.encode("latin-1") 
        ) 

    body = ( 
        request_body 
        if isinstance(request_body, str) 
        else request_body.decode("latin-1") 
    ) 

    post_data = urllib.parse.parse_qs(body) 
    
# verify we have our necessary arguments 
    ref = post_data.get("ref") or [] 
    if not ref: 
        return "error-ref" 

    what = post_data.get("what") or [] 
    if not what: 
        return "error-what" 

    what = what[0] 
    now = datetime.datetime.now(datetime.timezone.utc) 
    now = now.replace(microsecond=0, tzinfo=None).isoformat() 
    now += "+0000" 

    # dispatch 
    if what == "start": 
        handle_start(conn, now, ref) 
        return "ok" 

    if what == "image": 
        return handle_image(conn, post_data, now, ref) 

    if what == "end": 
        return handle_end(conn, post_data, now, ref) 

    return "error-unknown-" + what[0]

This first function processes your request data from raw bytes, ensuring that you have a dictionary-like object to get arguments from, then verifying that specific request arguments exist. Then finally, handle_request() dispatches to relevant sub-functions we are going to define. 

We do not provide a method to handle the “qrcode” event, as that may be days in advance of any image scanning, and in some cases a QR code may be generated and never used. Either way, you may consider adding a clause to the above dispatch function and enabling the webhook if you generate QR codes immediately before shipping, and you want to ensure you have records of when a shipment is about to leave your facility. 

Webhook for handle_start() 

Let’s look at our first webhook handler, handle_start(conn, now, ref): 

def handle_start(conn, now: str, ref: str): 
 exist_query = "SELECT pro FROM known_shipments WHERE pro = ?" 
 if list(conn.execute(exist_query, [ref])): 
   # don't re-insert 
   return 

 insert_data( 
    conn, 
    "known_shipments", 
    { 
      "pro": ref, 
      "status": "QR_SCANNED", 
      "longstatus": "QR Code was scanned", 
      "delivery_time": now, 
      }, 
    )

This function mostly ensures that there exists a row in known_shipments, and if not, that it at least has the status of “delivering”, as the delivery qr code was at least scanned. 

Webhook for handle_end() 

Handling the end of a shipment is very similar to the start of a shipment. Additionally, we now have more details about the final shipment status, which we include in handle_end(post_data, now, ref): 

def handle_end(conn, post_data: dict, now: str, ref: str) -> str:
 if not post_data.get("longstatus"): 
    return "error-longstatus" 

    # this matches what is returned by get_status() 
    status, _, longstatus = longstatus.partition(" - ") 

    # ensure the row exists 
    handle_start(conn, now, ref) 

    # update the row 
    conn.execute( 
        """ 
        UPDATE known_shipments 
        SET status = ?, longstatus = ? 
        WHERE pro = ? 
        """, 
        [status, longstatus, ref], 
    ) 
    conn.commit() 
    return "ok" 

In this case, we see our call to handle_start() ensures our known_shipments row exists, before we update it with the webhook-provided delivery status. 

Webhook for handle_image() 

Now that we’ve handled image upload start and end, collecting the images is all that remains from our original goal. 

def handle_image(conn, post_data: dict, now: str, ref: str) -> str:
for it in ("filename", "image"): 
        if not post_data.get(it): 
            return f"error-{it}" 

    image = post_data["image"][0] 
    if not image.startswith("data:image/jpeg;base64,"): 
        return "error-image-data" 

    try: 
        image_bytes = base64.b64decode(image.partition(",")[-1].encode()) 
    except Exception: 
        # POST body should support at least up to 418,000 bytes to 
        # receive image data. If your platform encodes base64 as handled 
        # by handle_request(), then you should ensure your post data limit 
        # is at least 558,000 bytes. 
        return "error-image-data" 

    # ensure foreign key exists 
    handle_start(conn, now, ref) 

    # will be <pro>_<image_type>.jpg, which matches the filenames returned 
    # by get_individual_images() 
    filename = post_data["filename"][0] 

    insert_data( 
        conn, 
        "shipment_images", 
        { 
            "known_pro": ref[0], 
            "image_identifier": filename, 
            "image_data": image_bytes, 
        }, 
    ) 
    return "ok" 

And we have arrived at our destination, handling individual images as they are received. If your web server limits the size of post requests, you may need to increase the size of the maximum post to at least 418,000 bytes or 558,000 bytes, depending on how your platform handles post requests. 

At this point we have functions to handle requesting status and image data to be downloaded to the local disk or stored in a database. We also have functions to handle incoming webhook requests for start shipment, shipment image, and end shipment events, again inserting into that database. 

For the sake of convenience and usability, we have provided all source code, along with additional documentation and comments on Github