Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def continuous_test_cases(value, expected):
return discrete_test_cases(value, expected) + [
(ComparisonOperator.GREATER_THAN, value, "gt", expected),
(ComparisonOperator.GREATER_THAN_OR_EQUAL_TO, value, "ge", expected),
(ComparisonOperator.LESS_THAN, value, "lt", expected),
(ComparisonOperator.LESS_THAN_OR_EQUAL_TO, value, "le", expected),
]
{"metrics": [METRIC], "params": [PARAM], "tags": [TAG]}
)
assert data == EXPERIMENT_RUN_DATA_BODY
def test_experiment_run_data_schema_empty():
data = ExperimentRunDataSchema().dump({})
assert data == {}
def test_experiment_run_data_schema_multiple():
data = ExperimentRunDataSchema().dump({"tags": [TAG, OTHER_TAG]})
assert data == {"tags": [TAG_BODY, OTHER_TAG_BODY]}
PROJECT_ID_FILTER = ProjectIdFilter(ComparisonOperator.EQUAL_TO, PROJECT_ID)
PROJECT_ID_FILTER_BODY = {
"by": "projectId",
"operator": "eq",
"value": str(PROJECT_ID),
}
TAG_FILTER = TagFilter("tag-key", ComparisonOperator.EQUAL_TO, "tag-value")
TAG_FILTER_BODY = {
"by": "tag",
"key": "tag-key",
"operator": "eq",
"value": "tag-value",
}
DEFINED_TEST_CASES = [
def continuous_test_cases(value, expected):
return discrete_test_cases(value, expected) + [
(ComparisonOperator.GREATER_THAN, value, "gt", expected),
(ComparisonOperator.GREATER_THAN_OR_EQUAL_TO, value, "ge", expected),
(ComparisonOperator.LESS_THAN, value, "lt", expected),
(ComparisonOperator.LESS_THAN_OR_EQUAL_TO, value, "le", expected),
]
def discrete_test_cases(value, expected):
return DEFINED_TEST_CASES + [
(ComparisonOperator.EQUAL_TO, value, "eq", expected),
(ComparisonOperator.NOT_EQUAL_TO, value, "ne", expected),
]
def discrete_test_cases(value, expected):
return DEFINED_TEST_CASES + [
(ComparisonOperator.EQUAL_TO, value, "eq", expected),
(ComparisonOperator.NOT_EQUAL_TO, value, "ne", expected),
]
ComparisonOperator.LESS_THAN,
ComparisonOperator.LESS_THAN_OR_EQUAL_TO,
],
)
def test_filter_schema_invalid_operator_no_key(filter_type, value, operator):
filter = filter_type(operator, value)
with pytest.raises(ValidationError, match="Not a discrete operator"):
FilterSchema().dump(filter)
"""
experiment_ids_filter = None
lifecycle_filter = None
filter = None
if experiment_ids is not None:
if len(experiment_ids) == 0:
return ListExperimentRunsResponse(
runs=[],
pagination=Pagination(
start=0, size=0, previous=None, next=None
),
)
experiment_id_filters = [
ExperimentIdFilter(ComparisonOperator.EQUAL_TO, experiment_id)
for experiment_id in experiment_ids
]
experiment_ids_filter = CompoundFilter(
LogicalOperator.OR, experiment_id_filters
)
if lifecycle_stage is not None:
lifecycle_filter = DeletedAtFilter(
ComparisonOperator.DEFINED,
lifecycle_stage == LifecycleStage.DELETED,
)
if experiment_ids_filter is not None and lifecycle_filter is not None:
filter = CompoundFilter(
LogicalOperator.AND, [experiment_ids_filter, lifecycle_filter]
)
elif experiment_ids_filter is not None:
def _serialize(self, value, attr, obj, **kwargs):
if obj.operator == ComparisonOperator.DEFINED:
field = fields.Boolean()
else:
field = self.other_field_type
return field._serialize(value, attr, obj, **kwargs)
DeleteExperimentRunsResponse
Containing lists of successfully deleted and conflicting (already
deleted) run IDs.
"""
endpoint = "/project/{}/run/delete/query".format(project_id)
if run_ids is None:
# Delete all runs in project
payload = {} # No filter
elif len(run_ids) == 0:
return DeleteExperimentRunsResponse(
deleted_run_ids=[], conflicted_run_ids=[]
)
else:
run_id_filters = [
RunIdFilter(ComparisonOperator.EQUAL_TO, run_id)
for run_id in run_ids
]
filter = CompoundFilter(LogicalOperator.OR, run_id_filters)
payload = {"filter": FilterSchema().dump(filter)}
return self._post(
endpoint, DeleteExperimentRunsResponseSchema(), json=payload
)
return obj
class _RunStatusFilterSchema(BaseSchema):
operator = EnumField(ComparisonOperator, by_value=True)
value = _FilterValueField(EnumField(ExperimentRunStatus, by_value=True))
by = fields.Constant("status", dump_only=True)
@pre_dump
def check_operator(self, obj):
_validate_discrete(obj.operator)
return obj
class _DeletedAtFilterSchema(BaseSchema):
operator = EnumField(ComparisonOperator, by_value=True)
value = _FilterValueField(fields.DateTime())
by = fields.Constant("deletedAt", dump_only=True)
class _TagFilterSchema(BaseSchema):
key = fields.String()
operator = EnumField(ComparisonOperator, by_value=True)
value = _FilterValueField(fields.String())
by = fields.Constant("tag", dump_only=True)
@pre_dump
def check_operator(self, obj):
_validate_discrete(obj.operator)
return obj