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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153 | class PyroJobGroupResource:
def __init__(self, client: PyroApiClient):
self.client = client
self._endpoint = "job_groups"
@classmethod
def from_client(cls, client: PyroApiClient) -> "PyroJobGroupResource":
return PyroJobGroupResource(client)
def create(self, name: str = "Untitled Group") -> PyroJobGroup:
"""
# Create a job group
## Example
```python
pyro = PyroDash(...)
group = pyro.job_groups.create("PY999")
```
"""
resp = self.client.request(POST, self._endpoint, {"name": name})
_dict = {**resp, "_resource": self}
return PyroJobGroup.from_dict(_dict)
def list_jobs(self, id: str) -> list[PyroJob]:
"""
# Retrieve jobs in a job group
## Example
```python
pyro = PyroDash(...)
job_group_id = "jg_2VF7h.."
jobs = pyro.job_groups.list_jobs(job_group_id)
```
"""
url = f"{self._endpoint}/{id}"
resp = self.client.request(GET, url)
jobs = []
for job in resp["jobs"]:
job_url = f"jobs/{job['id']}"
full_job_raw = self.client.request(GET, job_url)
as_dict = {**full_job_raw, "_resource": PyroJobResource(self.client)}
jobs.append(PyroJob.from_dict(as_dict))
return jobs
def add_job(self, id: str, job_id: str) -> PyroJob:
"""
# Add a job to a job group
## Example
```python
pyro = PyroDash(...)
group = pyro.job_groups.create()
job_id = "j_r62X.."
job = group.add_job(job_id)
```
"""
url = f"{self._endpoint}/{id}/add_job"
resp = self.client.request(POST, url, {"job_id": job_id})
_dict = {**resp, "_resource": PyroJobResource(self.client)}
return PyroJob.from_dict(_dict)
def create_job(self, id: str, job: Union[PyroJob, dict]) -> PyroJob:
"""
# Create and add a new job to a job group
This creates a new job and adds it to the specified job group.
Note: This does **not** copy the job's config.
Use `duplicate()` if you want to fully clone a job, or call `update()` after creation.
## Example
```python
pyro = PyroDash(...)
group = pyro.job_groups.get("g_123")
# From dict
job = group.create_job({"type": "fsim", "name": "Test", "description": "A test job"})
# From existing job
source_job = pyro.jobs.get("j_r62X..")
job = group.create_job(source_job)
```
"""
if isinstance(job, PyroJob):
return self.create_job_from_pyrojob(id, job)
elif isinstance(job, dict):
return self.create_job_from_dict(id, job)
else:
msg = "source job provided is of unexpected type. cannot create a new job from it."
raise TypeError(msg)
def create_job_from_dict(self, id: str, job: dict) -> PyroJob:
"""
# Create a job from a dictionary and add it to a job group
The dict must include a `"type"` field. Config is not copied.
## Example
```python
pyro = PyroDash(...)
group = pyro.job_groups.get("g_123")
job = group.create_job_from_dict({
"type": "fsim",
"name": "My Job",
"description": "Something useful"
})
```
"""
if job["type"] is None:
raise ValueError("dict specified does not have a type!")
job_rsc = PyroJobResource(self.client)
new_job = job_rsc.create(job["type"])
new_job.update(
name=job.get("name", "Unlabeled"), description=job.get("description", "")
)
self.add_job(id, new_job.id)
return job_rsc.get(new_job.id)
def create_job_from_pyrojob(self, id: str, job: PyroJob) -> PyroJob:
"""
# Create a job from another PyroJob and add it to a job group
Only the job's name and description are copied—not config.
## Example
```python
pyro = PyroDash(...)
group = pyro.job_groups.get("g_123")
template_job = pyro.jobs.get("j_r62X..")
new_job = group.create_job_from_pyrojob(template_job)
```
"""
if job.type is None:
raise ValueError("job specified does not have a type!")
job_rsc = PyroJobResource(self.client)
new_job = job_rsc.create(job.type) # pyright: ignore
new_job.update(name=job.name, description=job.description)
self.add_job(id, new_job.id)
return job_rsc.get(new_job.id)
|