Snowflake SQLAlchemy - Dynamically created column with Timestamp?

Question

This is a follow-up question to my previous one.

Snowflake SQLAlchemy - Create table with Timestamp?

I am dynamically creating columns and I have this schema.

I need the "my_time_stamp" column to have the server default.

json_cls_schema = {
    "clsname": "MyClass",
    "tablename": "my_table",
    "columns": [
        {"name": "id", "type": "integer", "is_pk": True, "is_auto" : True},
        {"name": "my_time_stamp", "type": "timestamp", 'serverdefault': text('current_timestamp')}
    ],
}

How would I set so that the code below generates this column.

my_time_stamp = Column(TIMESTAMP,server_default=text('current_timestamp()'))

Right now I am getting the error TypeError: Object of type TextClause is not JSON serializable and the dictionary I am generating this column from looks like this.

{'name': 'my_time_stamp', 'type': 'timestamp', 'serverdefault': <sqlalchemy.sql.elements.TextClause object at 0x000001B57664F560>}

code:

_type_lookup = {
    "integer": Integer,
    "timestamp": TIMESTAMP,


}

def mapping_for_json(json_cls_schema):
    clsdict = {"__tablename__": json_cls_schema["tablename"]}

    clsdict.update(
        {
            rec["name"]:
                Column(
                    _type_lookup[rec["type"]], primary_key=rec.get("is_pk", False),
                    autoincrement=rec.get("is_auto", False), server_default=rec.get("serverdefault", '')
                )
            for rec in json_cls_schema["columns"]
        }
    )


    return type(json_cls_schema["clsname"], (Base,), clsdict)

Answer 1

The (implicit) problem here is that json_cls_schema, as the name suggests, can be expected to be serialised to and deserialised from JSON, but sqlalchemy.text('CURRENT_TIMESTAMP') cannot be directly serialised to JSON.

The general solution for unserialisable objects is to define a custom function that will serialise the object. When it comes to deserialisation. the set of objects that we might encounter in a table definition is large, but finite, so object hooks provide a solution. All this is described in the docs for the json module in the standard library.

import json
from pprint import pprint

import sqlalchemy as sa


def default(o):
    """Serialise TextClause objects to JSON."""
    if isinstance(o, sa.TextClause):
        # Use a unique key to allow the hook to identify TextClauses.
        return {
            '__text__': {
                'text': o.text,
            }
        }
    raise TypeError(f'Object of type {type(o.__name__)} is not JSON serializable')


def as_text(dict_):
    """Identify serialised TextClause objects and deserialise them."""
    if '__text__' in dict_:
        text = dict_['__text__']['text']
        return sa.text(text)
    return dict_


# Pass the serialisation function to json.dump(s).
serialised_schema = json.dumps(json_cls_schema, default=default)

# Pass the deserialisation object hook function to json.load(s).
deserialised_schema = json.loads(serialised_schema, object_hook=as_text)
pprint(deserialised_schema)

Output

{'clsname': 'MyClass',
 'columns': [{'is_auto': True, 'is_pk': True, 'name': 'id', 'type': 'integer'},
             {'name': 'my_time_stamp',
              'serverdefault': <sqlalchemy.sql.elements.TextClause object at 0x7678facc6f90>,
              'type': 'timestamp'}],
 'tablename': 'my_table'}