Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"type": "javascript",
"name": "postagg_name",
"fieldNames": ["field1", "field2"],
"function": function,
}
postagg = get_post_agg(conf)
self.assertTrue(isinstance(postagg, models.JavascriptPostAggregator))
self.assertEqual(postagg.name, "postagg_name")
self.assertEqual(postagg.post_aggregator["type"], "javascript")
self.assertEqual(postagg.post_aggregator["fieldNames"], ["field1", "field2"])
self.assertEqual(postagg.post_aggregator["name"], "postagg_name")
self.assertEqual(postagg.post_aggregator["function"], function)
# Quantile
conf = {"type": "quantile", "name": "postagg_name", "probability": "0.5"}
postagg = get_post_agg(conf)
self.assertTrue(isinstance(postagg, postaggs.Quantile))
self.assertEqual(postagg.name, "postagg_name")
self.assertEqual(postagg.post_aggregator["probability"], "0.5")
# Quantiles
conf = {
"type": "quantiles",
"name": "postagg_name",
"probabilities": "0.4,0.5,0.6",
}
postagg = get_post_agg(conf)
self.assertTrue(isinstance(postagg, postaggs.Quantiles))
self.assertEqual(postagg.name, "postagg_name")
self.assertEqual(postagg.post_aggregator["probabilities"], "0.4,0.5,0.6")
# FieldAccess
conf = {"type": "fieldAccess", "name": "field_name"}
postagg = get_post_agg(conf)
self.assertTrue(isinstance(postagg, postaggs.Field))
for metric_name in metrics:
metric = metrics_dict[metric_name]
if metric.metric_type != 'postagg':
all_metrics.append(metric_name)
else:
conf = metric.json_obj
all_metrics += recursive_get_fields(conf)
all_metrics += conf.get('fieldNames', [])
if conf.get('type') == 'javascript':
post_aggs[metric_name] = JavascriptPostAggregator(
name=conf.get('name', ''),
field_names=conf.get('fieldNames', []),
function=conf.get('function', ''))
elif conf.get('type') == 'quantile':
post_aggs[metric_name] = Quantile(
conf.get('name', ''),
conf.get('probability', ''),
)
elif conf.get('type') == 'quantiles':
post_aggs[metric_name] = Quantiles(
conf.get('name', ''),
conf.get('probabilities', ''),
)
elif conf.get('type') == 'fieldAccess':
post_aggs[metric_name] = Field(conf.get('name'), '')
elif conf.get('type') == 'constant':
post_aggs[metric_name] = Const(
conf.get('value'),
output_name=conf.get('name', '')
)
elif conf.get('type') == 'hyperUniqueCardinality':
def get_post_agg(mconf: Dict) -> "Postaggregator":
"""
For a metric specified as `postagg` returns the
kind of post aggregation for pydruid.
"""
if mconf.get("type") == "javascript":
return JavascriptPostAggregator(
name=mconf.get("name", ""),
field_names=mconf.get("fieldNames", []),
function=mconf.get("function", ""),
)
elif mconf.get("type") == "quantile":
return Quantile(mconf.get("name", ""), mconf.get("probability", ""))
elif mconf.get("type") == "quantiles":
return Quantiles(mconf.get("name", ""), mconf.get("probabilities", ""))
elif mconf.get("type") == "fieldAccess":
return Field(mconf.get("name"))
elif mconf.get("type") == "constant":
return Const(mconf.get("value"), output_name=mconf.get("name", ""))
elif mconf.get("type") == "hyperUniqueCardinality":
return HyperUniqueCardinality(mconf.get("name"))
elif mconf.get("type") == "arithmetic":
return Postaggregator(
mconf.get("fn", "/"), mconf.get("fields", []), mconf.get("name", "")
)
else:
return CustomPostAggregator(mconf.get("name", ""), mconf)