Skip to content

Experiment manager

Experiment dataclass #

Experiment(id: str, name: str, state: str)

A created, activated experiment.

ExperimentManager #

ExperimentManager(openapi_service: OpenAPIService)

Internal: creates attached-scope 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 
35
36
37
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,
    user_enrollment: ExperimentUserEnrollment = STICKY,
    description: str = ""
) -> Experiment

Creates and activates an attached-scope experiment.

The experiment is created in the draft state and immediately activated, so it is already serving by the time this returns — meant to be paired with 🇵🇾meth:run_experiment, which attaches it to the job's campaign before that campaign can serve.

Source code in src/rapidata/rapidata_client/experiment/experiment_manager.py 
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def create(
    self,
    *,
    name: str,
    treatment_bps: int = 5000,
    flags: list[FeatureFlag] | None = None,
    user_enrollment: ExperimentUserEnrollment = ExperimentUserEnrollment.STICKY,
    description: str = "",
) -> Experiment:
    """Creates and activates an attached-scope experiment.

    The experiment is created in the draft state and immediately activated, so
    it is already serving by the time this returns — meant to be paired with
    :py:meth:`run_experiment`, which attaches it to the job's campaign before
    that campaign can serve.
    """
    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,
        )
        from rapidata.api_client.models.change_experiment_state_endpoint_input import (
            ChangeExperimentStateEndpointInput,
        )
        from rapidata.api_client.models.change_experiment_state_endpoint_state_action import (
            ChangeExperimentStateEndpointStateAction,
        )
        from rapidata.api_client.models.experiment_scope import ExperimentScope

        api = self._openapi_service.campaign.experiment_api
        created = api.campaign_experiments_post(
            create_experiment_endpoint_input=CreateExperimentEndpointInput(
                name=name,
                description=description,
                scope=ExperimentScope.ATTACHED,
                flags=flags or [],
                userEnrollment=user_enrollment,
                split=CreateExperimentEndpointSplit(treatmentBps=treatment_bps),
            )
        )
        activated = api.campaign_experiments_experiment_id_state_post(
            created.id,
            ChangeExperimentStateEndpointInput(
                action=ChangeExperimentStateEndpointStateAction.ACTIVATE
            ),
        )
        logger.info("Created and activated experiment '%s'", activated.id)
        return Experiment(id=activated.id, name=activated.name, state="active")

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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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
        )