sql server compiles temporal clauses

sa.py

# render `period for system_time` clause for mssql temporal tables
# example (specifically pay attention to the info argument passed to sys_end_time)
# the render_period has to reference the column names used to track the system start and end time
# class Card(Base):
#     sys_start_time = Column(
#         DATETIME2,
#         Computed(text("ROW START HIDDEN")),
#         nullable=False,
#         info={"is_temporal": True},
#     )
#     sys_end_time = Column(
#         DATETIME2,
#         Computed(text("ROW END HIDDEN")),
#         nullable=False,
#         info={
#             "is_temporal": True,
#             "render_period": "PERIOD FOR SYSTEM_TIME (sys_start_time, sys_end_time)",
#         },
#     )
#     __table_args__ = {"info": {"history_table": "CardHistory"}}
@compiles(schema.CreateColumn, "mssql")
def render_period(element, compiler, **kw):
    column = element.element

    # if render period then render the column and add the period afterwards
    if column.computed is not None and "render_period" in column.info:
        text = compiler.visit_create_column(element, **kw)
        # https://github.com/sqlalchemy/sqlalchemy/blob/03989d1dce80999bb9ea1a7d36df3285e5ce4c3b/lib/sqlalchemy/sql/compiler.py#L4574-L4576
        text += ", \n"
        text += "\t"
        text += column.info["render_period"]
        return text
    else:
        return compiler.visit_create_column(element, **kw)


# render `WITH (SYSTEM_VERSIONING = ON (HISTORY_TABLE` clause for msslq temporal tables
# example (specifically pay attention to the info argument passed in __table_args__)
# it is used to pass the name of the history table
# class Card(Base):
#     sys_start_time = Column(
#         DATETIME2,
#         Computed(text("ROW START HIDDEN")),
#         nullable=False,
#         info={"is_temporal": True},
#     )
#     sys_end_time = Column(
#         DATETIME2,
#         Computed(text("ROW END HIDDEN")),
#         nullable=False,
#         info={
#             "is_temporal": True,
#             "render_period": "PERIOD FOR SYSTEM_TIME (sys_start_time, sys_end_time)",
#         },
#     )
#     __table_args__ = {"info": {"history_table": "CardHistory"}}
@compiles(schema.CreateTable, "mssql")
def compile_temporal_table(element, compiler, **kw):
    table = element.element

    if "history_table" in table.info:
        text: str = compiler.visit_create_table(element, **kw)
        # see https://github.com/sqlalchemy/sqlalchemy/blob/03989d1dce80999bb9ea1a7d36df3285e5ce4c3b/lib/sqlalchemy/sql/compiler.py#L4592
        # we assume post_create_table is empty and it is for mssql
        text.rstrip()
        text += f"\n)\nWITH (SYSTEM_VERSIONING = ON (HISTORY_TABLE = {table.info['history_table']}))\n\n"
        return text
    else:
        return compiler.visit_create_table(element, **kw)


# render `GENERATED ALWAYS AS ROW START` clause for msslq temporal tables
# example (specifically pay attention to the info argument passed in column args)
# is_temporal is used to make the column generate correctly
# class Card(Base):
#     sys_start_time = Column(
#         DATETIME2,
#         Computed(text("ROW START HIDDEN")),
#         nullable=False,
#         info={"is_temporal": True},
#     )
#     sys_end_time = Column(
#         DATETIME2,
#         Computed(text("ROW END HIDDEN")),
#         nullable=False,
#         info={
#             "is_temporal": True,
#             "render_period": "PERIOD FOR SYSTEM_TIME (sys_start_time, sys_end_time)",
#         },
#     )
#     __table_args__ = {"info": {"history_table": "CardHistory"}}
@compiles(schema.Computed, "mssql")
def visit_computed_column(element, compiler, **kw):
    if "is_temporal" in element.info and element.info["is_temporal"] == True:
        return "GENERATED ALWAYS AS (%s)" % compiler.sql_compiler.process(
            element.sqltext, include_table=False, literal_binds=True
        )
    else:
        return compiler.visit_computed_column(element, **kw)