in experimenter/experimenter/jetstream/models.py [0:0]
def create_results_object_model(data: JetstreamData):
branches = {data_point.branch for data_point in data}
# create a dynamic model with all branches, leveraging BranchComparisonData
PairwiseBranchComparisonData = create_model(
"PairwiseBranchComparisonData",
**{
branch: (BranchComparisonData, Field(default_factory=BranchComparisonData))
for branch in branches
},
)
# create a dynamic model with all branches, leveraging SignificanceData
PairwiseSignificanceData = create_model(
"PairwiseSignificanceData",
**{
branch: (SignificanceData, Field(default_factory=SignificanceData))
for branch in branches
},
)
class PairwiseMetricData(MetricData):
difference: PairwiseBranchComparisonData = Field(
default_factory=PairwiseBranchComparisonData
)
relative_uplift: PairwiseBranchComparisonData = Field(
default_factory=PairwiseBranchComparisonData
)
significance: PairwiseSignificanceData = Field(
default_factory=PairwiseSignificanceData
)
metrics = {data_point.metric for data_point in data}
# Dynamically create our grouped models which are dependent on metrics
# available for a given experiment
SearchData = create_model(
"SearchData",
**{
metric_name: (PairwiseMetricData, Field(default_factory=PairwiseMetricData))
for metric_name in metrics
if metric_name in SEARCH_METRICS
},
)
UsageData = create_model(
"UsageData",
**{
metric_name: (PairwiseMetricData, Field(default_factory=PairwiseMetricData))
for metric_name in metrics
if metric_name in USAGE_METRICS
},
)
OtherData = create_model(
"OtherData",
**{
metric_name: (PairwiseMetricData, Field(default_factory=PairwiseMetricData))
for metric_name in metrics
if metric_name not in SEARCH_METRICS + USAGE_METRICS
},
)
class BranchData(BaseModel):
search_metrics: SearchData = Field(default_factory=SearchData)
usage_metrics: UsageData = Field(default_factory=UsageData)
other_metrics: OtherData = Field(default_factory=OtherData)
class Branch(BaseModel):
is_control: bool = False
branch_data: BranchData = Field(default_factory=BranchData)
# Create ResultsObjectModel model which is dependent on
# branches available for a given experiment
return create_model(
"ResultsObjectModel",
**{branch: (Branch, Field(default_factory=Branch)) for branch in branches},
__base__=ResultsObjectModelBase,
)