dataclasses.py 5.01 KB
Newer Older
Anthony Leung's avatar
Anthony Leung committed
1
from dataclasses import dataclass, field
2
import datetime
Anthony Leung's avatar
Anthony Leung committed
3
from typing import Any, List, Dict
4
5
6
7
8
9
10
11
12
import uuid


@dataclass
class Job:
    name: str
    description: str
    uuid: uuid.UUID
    short_uuid: str
13
    project: dict
14
    notifications: dict
15
    schedules: list
16
17
18
    created: datetime.datetime
    modified: datetime.datetime
    environment: str
19
    timezone: str
20
21


22
23
24
25
26
27
28
29
@dataclass
class Project:
    name: str
    description: str
    uuid: uuid.UUID
    short_uuid: str
    created_by: dict
    package: dict
30
    notifications: dict
31
    template: str
32
33
    permission: dict
    visibility: str
34
35
36
    created: datetime.datetime
    modified: datetime.datetime
    workspace: dict
37
    is_member: bool
38
    url: str = None
39
40

    def __str__(self):
Anthony Leung's avatar
Anthony Leung committed
41
        return f"{self.name} {self.short_uuid}"
42
43


44
45
@dataclass
class Run:
46
47
    """
    This dataclass holds information about the run which just started
Robbert Bos's avatar
Robbert Bos committed
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
    """

    message_type: str
    status: str
    uuid: uuid.UUID
    short_uuid: str
    name: str
    next_url: str
    job: dict
    workspace: dict
    project: dict
    created: datetime.datetime
    updated: datetime.datetime
    finished: datetime.datetime = None
    duration: int = 0


@dataclass
class RunStatus:
    """
    This dataclass holds information about the status of a run.
69
70
    """

71
72
73
74
    message_type: str
    status: str
    uuid: uuid.UUID
    short_uuid: str
75
    name: str
76
77
78
79
    next_url: str
    job: dict
    workspace: dict
    project: dict
80
    environment: dict
81
82
83
    created: datetime.datetime
    updated: datetime.datetime
    finished: datetime.datetime = None
84
    duration: int = 0
85
86


Anthony Leung's avatar
Anthony Leung committed
87
88
@dataclass
class RunInfo:
89
    name: str
Anthony Leung's avatar
Anthony Leung committed
90
91
92
93
    description: str
    status: str
    uuid: uuid.UUID
    short_uuid: str
94

Anthony Leung's avatar
Anthony Leung committed
95
96
97
98
99
100
101
    project: dict
    artifact: dict
    package: dict
    owner: dict
    trigger: dict
    payload: dict
    jobdef: dict
102
103
    environment: dict

Anthony Leung's avatar
Anthony Leung committed
104
    metricsmeta: dict
Anthony Leung's avatar
Anthony Leung committed
105
    variablesmeta: dict
106

Anthony Leung's avatar
Anthony Leung committed
107
108
    created: datetime.datetime
    modified: datetime.datetime
109
110
111
112
    duration: int

    result: dict = None
    started: datetime.datetime = None
Anthony Leung's avatar
Anthony Leung committed
113
114
115
    finished: datetime.datetime = None

    metrics = []
Anthony Leung's avatar
Anthony Leung committed
116
    variables = []
Anthony Leung's avatar
Anthony Leung committed
117
118


119
120
121
122
123
124
125
126
127
128
@dataclass
class User:
    short_uuid: str
    name: str
    email: str
    is_active: bool
    date_joined: datetime.datetime
    last_login: datetime.datetime


129
130
131
132
133
134
135
136
137
138
139
140
141
142
@dataclass
class Variable:
    name: str
    value: str
    short_uuid: str
    uuid: uuid.UUID
    is_masked: bool
    created: datetime.datetime
    modified: datetime.datetime
    project: dict


@dataclass
class Workspace:
143
    name: str
144
145
146
    description: str
    uuid: uuid.UUID
    short_uuid: str
147
148
    visibility: str
    is_member: bool
149
150
151
152
    created: datetime.datetime
    modified: datetime.datetime

    def __str__(self):
153
        return f"{self.name} {self.short_uuid}"
Anthony Leung's avatar
Anthony Leung committed
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196


@dataclass
class MetricDataPair:
    name: str
    value: Any
    dtype: str

    def to_dict(self, yes=True) -> Dict:
        return {"name": self.name, "value": self.value, "type": self.dtype}


@dataclass
class MetricLabel:
    name: str
    value: Any
    dtype: str

    def to_dict(self, yes=True) -> Dict:
        return {"name": self.name, "value": self.value, "type": self.dtype}


@dataclass
class Metric:
    metric: MetricDataPair
    label: List[MetricLabel] = field(default_factory=list)
    run_suuid: str = None
    created: datetime.datetime = None

    def __post_init__(self):
        if not self.created:
            # record the created time always in UTC time so that we don't have
            # to figure out what local timezone is, this is always correct
            # and we don't have to calculate back to utc
            self.created = datetime.datetime.now(datetime.timezone.utc)

    def to_dict(self, yes=True) -> Dict:
        return {
            "metric": self.metric.to_dict(),
            "label": [label.to_dict() for label in self.label],
            "run_suuid": self.run_suuid,
            "created": self.created,
        }
Anthony Leung's avatar
Anthony Leung committed
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239


@dataclass
class VariableDataPair:
    name: str
    value: Any
    dtype: str

    def to_dict(self, yes=True) -> Dict:
        return {"name": self.name, "value": self.value, "type": self.dtype}


@dataclass
class VariableLabel:
    name: str
    value: Any
    dtype: str

    def to_dict(self, yes=True) -> Dict:
        return {"name": self.name, "value": self.value, "type": self.dtype}


@dataclass
class VariableTracked:
    variable: VariableDataPair
    label: List[VariableLabel] = field(default_factory=list)
    run_suuid: str = None
    created: datetime.datetime = None

    def __post_init__(self):
        if not self.created:
            # record the created time always in UTC time so that we don't have
            # to figure out what local timezone is, this is always correct
            # and we don't have to calculate back to utc
            self.created = datetime.datetime.now(datetime.timezone.utc)

    def to_dict(self, yes=True) -> Dict:
        return {
            "variable": self.variable.to_dict(),
            "label": [label.to_dict() for label in self.label],
            "run_suuid": self.run_suuid,
            "created": self.created,
        }