Skip to content

Experiment manager

Experiment dataclass #

Experiment(id: str, name: str)

A created experiment.

ExperimentManager #

ExperimentManager(openapi_service: OpenAPIService)

Internal: creates experiments and launches jobs under them.

Not part of the public SDK surface — Rapidata-internal, undocumented.

Source code in src/rapidata/rapidata_client/experiment/experiment_manager.py 
31
32
33
def __init__(self, openapi_service: OpenAPIService):
    self._openapi_service = openapi_service
    logger.debug("ExperimentManager initialized")

create #

create(
    *,
    name: str,
    treatment_bps: int = 5000,
    flags: list[FeatureFlag] | None = None,
    description: str = ""
) -> Experiment

Creates an experiment.

It is live as soon as a campaign references it — meant to be paired with 🇵🇾meth:run_experiment, which attaches it to the job's campaign.

Source code in src/rapidata/rapidata_client/experiment/experiment_manager.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def create(
    self,
    *,
    name: str,
    treatment_bps: int = 5000,
    flags: list[FeatureFlag] | None = None,
    description: str = "",
) -> Experiment:
    """Creates an experiment.

    It is live as soon as a campaign references it — meant to be paired with
    :py:meth:`run_experiment`, which attaches it to the job's campaign.
    """
    with tracer.start_as_current_span("ExperimentManager.create"):
        from rapidata.api_client.models.create_experiment_endpoint_input import (
            CreateExperimentEndpointInput,
        )
        from rapidata.api_client.models.create_experiment_endpoint_split import (
            CreateExperimentEndpointSplit,
        )

        api = self._openapi_service.campaign.experiment_api
        created = api.campaign_experiments_post(
            create_experiment_endpoint_input=CreateExperimentEndpointInput(
                name=name,
                description=description,
                flags=flags or [],
                split=CreateExperimentEndpointSplit(treatmentBps=treatment_bps),
            )
        )
        logger.info("Created experiment '%s'", created.id)
        return Experiment(id=created.id, name=created.name)

run_experiment #

run_experiment(config: ExperimentConfig) -> RapidataJob

Launches config.job_definition on config.audience under the experiment.

Raises:

Type Description
ValueError

If config.experiment_id is empty.
Source code in src/rapidata/rapidata_client/experiment/experiment_manager.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
def run_experiment(self, config: ExperimentConfig) -> RapidataJob:
    """Launches ``config.job_definition`` on ``config.audience`` under the experiment.

    Raises:
        ValueError: If ``config.experiment_id`` is empty.
    """
    with tracer.start_as_current_span("ExperimentManager.run_experiment"):
        if not config.experiment_id:
            raise ValueError("ExperimentConfig.experiment_id must not be empty")

        return config.audience._create_job(
            config.job_definition, experiment_id=config.experiment_id
        )