def sampled_exposure_signal()

in jetstream/preview.py [0:0]

• 26 lines of code
• 5 McCabe index (conditional complexity)

def sampled_exposure_signal(start_date, config, population_sample_size) -> ExposureSignalDefinition:
    enrollments_query_type = PLATFORM_CONFIGS[config.experiment.app_name].enrollments_query_type

    # add sampling and remove matching on experiment slug (because no clients enrolled) for exposure
    if enrollments_query_type == EnrollmentsQueryType.NORMANDY:
        exposure_signal = ExposureSignalDefinition(
            name="sampled_preview_exposure",
            data_source=DataSourceReference(name="events"),
            select_expression=f"MOD(event_timestamp) = 0 AND sample_id < {population_sample_size}",
            description="Sampled Exposure Signal for Preview",
            friendly_name="Sampled Exposure Signal for Preview",
        )
    elif enrollments_query_type in [
        EnrollmentsQueryType.GLEAN_EVENT,
        EnrollmentsQueryType.FENIX_FALLBACK,
    ]:
        exposure_signal = ExposureSignalDefinition(
            name="sampled_preview_exposure",
            data_source=DataSourceReference(name="events"),
            select_expression=f"sample_id < {population_sample_size}",
            description="Sampled Exposure Signal for Preview",
            friendly_name="Sampled Exposure Signal for Preview",
        )
    else:
        raise ValueError(
            f"Cannot generate exposures for enrollment query type '{enrollments_query_type}'"
        )

    return exposure_signal