A simple example for a SQLIte pod for Babashka

pod-bb-python-sqlite.py

import json
import sqlite3
import sys

from bcoding import bencode, bdecode


def read():
    return dict(bdecode(sys.stdin.buffer))


def write(obj):
    sys.stdout.buffer.write(bencode(obj))
    sys.stdout.flush()


def debug(msg):
    with open("debug.log", "a") as f:
        f.write(str(msg) + "\n")


def main():
    while True:
        msg = read()

        op = msg["op"]

        if op == "describe":
            write(
                {
                    "format": "json",
                    "vars": [{"namespace": "pod.sqlite", "name": "execute!"}],
                }
            )
        elif op == "invoke":
            var = msg["var"]
            id = msg["id"]
            args = json.loads(msg["args"])

            conn = sqlite3.connect("bb.db")
            c = conn.cursor()

            result = None

            if var == "pod.sqlite/execute!":
                try:
                    result = c.execute(*args)
                except Exception as e:
                    debug(e)

            write({"value": json.dumps(result.fetchall()), "id": id, "status": ["done"]})

            conn.commit()
            conn.close()


if __name__ == "__main__":
    main()