def _publish_view()

in jetstream/analysis.py [0:0]

• 19 lines of code
• 3 McCabe index (conditional complexity)

    def _publish_view(self, window_period: AnalysisPeriod, table_prefix=None, analysis_basis=None):
        assert self.config.experiment.normandy_slug is not None
        normalized_slug = bq_normalize_name(self.config.experiment.normandy_slug)
        view_name = "_".join([normalized_slug, window_period.table_suffix])
        wildcard_expr = "_".join([normalized_slug, window_period.value, "*"])

        if analysis_basis:
            normalized_postfix = bq_normalize_name(analysis_basis)
            view_name = "_".join([normalized_slug, normalized_postfix, window_period.table_suffix])
            wildcard_expr = "_".join(
                [normalized_slug, normalized_postfix, window_period.value, "*"]
            )

        if table_prefix:
            normalized_prefix = bq_normalize_name(table_prefix)
            view_name = "_".join([normalized_prefix, view_name])
            wildcard_expr = "_".join([normalized_prefix, wildcard_expr])

        sql = dedent(
            f"""
            CREATE OR REPLACE VIEW `{self.project}.{self.dataset}.{view_name}` AS (
                SELECT
                    *,
                    CAST(_TABLE_SUFFIX AS int64) AS window_index
                FROM `{self.project}.{self.dataset}.{wildcard_expr}`
            )
            """
        )
        self.bigquery.execute(sql)