sqlite_in_memory_func.py

import dill
import sqlite3


def in_memory_serialization(obj, op):

    if op == 'serialize':
        try:
            return dill.dumps(obj)
        except:
            # if we fail, then pickle it by reference
            # see issue: https://github.com/uqfoundation/dill/issues/128
            return dill.dumps(obj, byref=True)
    elif op == 'deserialize':
        return dill.loads(obj)  # Deserialize from the byte stream
    else:
        raise Exception('Unsupported Operation')



def setup_db_and_inject_func(func_obj:callable, func_tag:str):
    # Pickle the function
    pickled_function = in_memory_serialization(func_obj, op='serialize')

    # Connect to the SQLite database (or create it if it doesn't exist)
    conn = sqlite3.connect('functions.db')
    c = conn.cursor()

    # Create a table to store the pickled functions
    c.execute('''CREATE TABLE IF NOT EXISTS functions (name TEXT, function BLOB)''')

    # Insert the pickled function into the table
    c.execute("INSERT INTO functions (name, function) VALUES (?, ?)", (func_tag, pickled_function))

    # Commit the transaction and close the connection
    conn.commit()
    conn.close()


def read_obj_from_db(func_tag:str):
    # Connect to the SQLite database
    conn = sqlite3.connect('functions.db')
    c = conn.cursor()

    # Retrieve the pickled function
    c.execute("SELECT function FROM functions WHERE name=?", (func_tag,))
    pickled_function = c.fetchone()[0]

    # Unpickle the function
    unpickled_function = in_memory_serialization(pickled_function, op='deserialize')

    # Close the connection
    conn.close()

    return unpickled_function


def function_multiply(x):
    return x * 2

setup_db_and_inject_func(func_obj=function_multiply, func_tag='myfunction')
func = read_obj_from_db(func_tag='myfunction')


func(100)