Dynamically generating SQLAlchemy tables

Sat 22 February 2020

Background

Here's the background: I had some structured data that I wanted to log to a database. That took the form of Python classes something like this:

class Message:
    code = 0x00
    endianness = '!'
    payload_format = {
        'device_address': 'L',
        'packet_id': 'L',
        'length': 'L',
        'enable': 'L',
    }

    def __init__(self, **kwargs):
        self.fields = kwargs

I wrote a module [struct-parse][struct-parse] sort of for this purpose a few weeks ago. It lets you parse the string format specified in the struct module.

Dynamically declaring classes

The first step was learning how to dynamically define classes in Python. Here's how you declare a class Foo that's a subclass of Bar. It has a class member called device_address that's initialized to None.

Foo = type('Foo', (Bar,) {'device_address': None})

Declaring SQLAlchemy tables

from sqlalchemy import Column, Integer, Float
from sqlalchemy.ext.declarative import declarative_base
import struct_parse as sp


Base = declarative_base()


_type_translation = {
    sp.FieldType.BOOL: Integer,
    sp.FieldType.CHAR: Integer,
    # sp.FieldType.CHAR_ARRAY: Integer,
    sp.FieldType.DOUBLE: Float,
    sp.FieldType.FLOAT: Float,
    sp.FieldType.HALF_PRECISION_FLOAT: Float,
    sp.FieldType.INT: Integer,
    sp.FieldType.LONG: Integer,
    sp.FieldType.LONG_LONG: Integer,
    sp.FieldType.PAD: Integer,
    sp.FieldType.SHORT: Integer,
    sp.FieldType.SIGNED_CHAR: Integer,
    sp.FieldType.SIZE_T: Integer,
    sp.FieldType.SSIZE_T: Integer,
    sp.FieldType.UNSIGNED_CHAR: Integer,
    sp.FieldType.UNSIGNED_INT: Integer,
    sp.FieldType.UNSIGNED_LONG: Integer,
    sp.FieldType.UNSIGNED_LONG_LONG: Integer,
    sp.FieldType.UNSIGNED_SHORT: Integer,
    # sp.FieldType.VOID_POINTER: Integer,
}


class_members = {
    '__tablename__': Message.__name__,
    'id': Column(Integer, primary_key=True),
}

table_classes = {}

for field_name, field_fmt in Message.payload_format.items():
    field_t = sp.parse(Message.endianness + field_fmt)
    class_members[field_name] = Column(_type_translation[field_t])
    TableClass = type('Message', (Base,), class_members)
    table_classes[Message.__name__] = TableClass

This declares a class that models a table to store Message objects. To actually create the database,

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///log.db')
Base.metadata.create_all(engine)

Now, let's insert a Message object into the Message table. Assume the Message structure definition is in the structures module, and the table is in the schema module.

import structures
import schema

Session = sessionmaker(bind=engine)
session = Session()

msg = structures.Message(
        device_address=0x42,
        packet_id=0x40,
        length=4,
        enable=0)

session.add(schema.table_classes[type(msg).__name__](**msg.fields))

We use type(msg).__name__ to look up the corresponding class that models the table in the schema module. Then, we expand the keyword arguments that msg was initialized with (notice they were stored in structures.Message.__init__) to set the fields of the table.

Done! And this approach seems to work.