code: def create_marker_table(self):
          
          if self.marker_table is None:
              self.marker_table = luigi.configuration.get_config().get('sqlalchemy', 'marker-table', 'table_updates')
  
          engine = self.engine
  
          with engine.begin() as con:
              metadata = sqlalchemy.MetaData()
              if not con.dialect.has_table(con, self.marker_table):
                  self.marker_table_bound = sqlalchemy.Table(
                      self.marker_table, metadata,
                      sqlalchemy.Column("update_id", sqlalchemy.String(128), primary_key=True),
                      sqlalchemy.Column("target_table", sqlalchemy.String(128)),
                      sqlalchemy.Column("inserted", sqlalchemy.DateTime, default=datetime.datetime.now()))
                  metadata.create_all(engine)
              else:
                  metadata.reflect(only=[self.marker_table], bind=engine)
                  self.marker_table_bound = metadata.tables[self.marker_table]
docs: Create marker table if it doesn't exist.
  
          Using a separate connection since the transaction might have to be reset.