Created
March 10, 2022 02:44
-
-
Save lukesmurray/c3339a6901a5fe890a1103e72aa45a30 to your computer and use it in GitHub Desktop.
sql server compiles temporal clauses
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment