def experiment_table_first_updated()

in jetstream/bigquery_client.py [0:0]

• 16 lines of code
• 6 McCabe index (conditional complexity)

    def experiment_table_first_updated(self, slug: str) -> datetime | None:
        """Get the timestamp for when an experiment related table was updated last."""
        if slug is None:
            return None

        table_prefix = bq_normalize_name(slug)

        job = self.client.query(
            rf"""
            SELECT
                table_name,
                REGEXP_EXTRACT_ALL(
                    option_value,
                    '.*STRUCT\\(\"last_updated\", \"([^\"]+)\"\\).*'
                ) AS last_updated
            FROM
            {self.dataset}.INFORMATION_SCHEMA.TABLE_OPTIONS
            WHERE option_name = 'labels' AND table_name LIKE "enrollments_{table_prefix}%"
            """
        )
        result = list(job.result())

        table_first_updated = None

        for row in result:
            if not len(row.last_updated):
                continue
            updated_timestamp = UTC.localize(datetime.utcfromtimestamp(int(row.last_updated[0])))

            if table_first_updated is None or updated_timestamp < table_first_updated:
                table_first_updated = updated_timestamp

        return table_first_updated