blob: 77cf28da0c5bc012e834e9add0f259876c26041a (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
"""API for interacting with Cedar."""
from typing import Union, Dict, Any, List
import requests
import yaml
class CedarRollup:
"""Representation of Cedar rollup."""
name: str
val: Union[int, float]
def __init__(self, json: Dict[str, Any]):
"""Initialize."""
self.name = json["name"]
self.val = json["val"]
class CedarPerfData:
"""Representation of Cedar performance data."""
test_name: str
thread_level: int
perf_rollups: List[CedarRollup]
def __init__(self, json: Dict[str, Any]):
"""Initialize."""
self.test_name = json["info"]["test_name"]
self.thread_level = json["info"]["args"]["thread_level"]
self.perf_rollups = [CedarRollup(rollup) for rollup in json["rollups"]["stats"]]
class CedarApi:
"""Representation of Cedar API."""
DEFAULT_API_SERVER = "https://cedar.mongodb.com"
def __init__(self, evg_api_config: str):
"""Initialize."""
with open(evg_api_config) as fh:
evg_config = yaml.safe_load(fh)
self.headers = {
"Api-User": evg_config["user"],
"Api-Key": evg_config["api_key"],
}
def get_perf_data_by_task_id(self, task_id: str) -> List[CedarPerfData]:
"""Get performance data by Evergreen task id."""
url = f"{self.DEFAULT_API_SERVER}/rest/v1/perf/task_id/{task_id}"
res = requests.get(url, headers=self.headers)
res.raise_for_status()
return [CedarPerfData(perf_data) for perf_data in res.json()]
|
