pythonsqlalchemysnowflake-cloud-data-platform

Snowflake SQLAlchemy - Create table with Timestamp?


I am creating table definitions with SqlAlchemy in Python. I am able to successfully create a table with primary key which autoincrements and some other fields. However, I haven't been able to find any documentation on how to add a field with a timestamp which will be populated when a new record is added. For reference, this is how you would create this field directly in a Snowflake worksheet.

CREATE OR REPLACE TABLE My_Table(
 TABLE_ID NUMBER NOT NULL PRIMARY KEY,
 ... Other fields
TIME_ADDED TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

This is my Python code...

from sqlalchemy import Column, DateTime, Integer, Text, String
from sqlalchemy.orm import declarative_base
from sqlalchemy import create_engine

Base = declarative_base()


class My_Table(Base):
    __tablename__ = 'my_table'

    TABLE_ID = Column(Integer, primary_key=True, autoincrement=True)
    .. other columns
    .. Also need to create TIME_ADDED field


engine = create_engine(
    'snowflake://{user}:{password}@{account_identifier}/{database}/{schema}?warehouse={warehouse}'.format(
        account_identifier=account,
        user=username,
        password=password,
        database=database,
        warehouse=warehouse,
        schema=schema
    )
)

Base.metadata.create_all(engine)

Has anyone done this and, if so, could you point me in the right direction?


Solution

  • It is fairly straightforward to add a creation timestamp using SQLAlchemy.

    from sqlalchemy import Column, TIMESTAMP, Integer, text
    from sqlalchemy.orm import declarative_base
    
    Base = declarative_base()
    
    
    class My_Table(Base):  # I'd suggest calling this MyTable within python.
        __tablename__ = 'my_table'
    
        TABLE_ID = Column(Integer, primary_key=True, autoincrement=True)
        # ... other columns
        TIME_ADDED = Column(TIMESTAMP, server_default=text('current_timestamp()'))  # See Note!
    

    This approach is discussed in more detail here, and in the SQLAlchemy docs

    Note: Just as a word of caution, I do not have snowflake installed on my machine but my understanding is that it should handle similarly to MySQL / MariaDB, for which the above method works. I could imagine snowflake requires a different current_timestamp() command internally, in which case you should replace it with a valid snowflake specific timestamp command.