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
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
|
from enum import Enum
from typing import Any, Optional
from moto.utilities.utils import get_partition
from .exceptions import ValidationError
def make_arn_for_compute_env(account_id: str, name: str, region_name: str) -> str:
return f"arn:{get_partition(region_name)}:batch:{region_name}:{account_id}:compute-environment/{name}"
def make_arn_for_job_queue(account_id: str, name: str, region_name: str) -> str:
return f"arn:{get_partition(region_name)}:batch:{region_name}:{account_id}:job-queue/{name}"
def make_arn_for_job(account_id: str, job_id: str, region_name: str) -> str:
return f"arn:{get_partition(region_name)}:batch:{region_name}:{account_id}:job/{job_id}"
def make_arn_for_task_def(
account_id: str, name: str, revision: int, region_name: str
) -> str:
return f"arn:{get_partition(region_name)}:batch:{region_name}:{account_id}:job-definition/{name}:{revision}"
def lowercase_first_key(some_dict: dict[str, Any]) -> dict[str, Any]:
new_dict: dict[str, Any] = {}
for key, value in some_dict.items():
new_key = key[0].lower() + key[1:]
try:
if isinstance(value, dict):
new_dict[new_key] = lowercase_first_key(value)
elif all(isinstance(v, dict) for v in value):
new_dict[new_key] = [lowercase_first_key(v) for v in value]
else:
new_dict[new_key] = value
except TypeError:
new_dict[new_key] = value
return new_dict
def validate_job_status(target_job_status: str, valid_job_statuses: list[str]) -> None:
if target_job_status not in valid_job_statuses:
raise ValidationError(
"1 validation error detected: Value at 'current_status' failed "
f"to satisfy constraint: Member must satisfy enum value set: {valid_job_statuses}"
)
class JobStatus(str, Enum):
SUBMITTED = "SUBMITTED"
PENDING = "PENDING"
RUNNABLE = "RUNNABLE"
STARTING = "STARTING"
RUNNING = "RUNNING"
SUCCEEDED = "SUCCEEDED"
FAILED = "FAILED"
@classmethod
def job_statuses(self) -> list[str]:
return sorted([item.value for item in JobStatus])
@classmethod
def is_job_already_started(self, current_status: str) -> bool:
validate_job_status(current_status, JobStatus.job_statuses())
return current_status not in [
JobStatus.SUBMITTED,
JobStatus.PENDING,
JobStatus.RUNNABLE,
JobStatus.STARTING,
]
@classmethod
def is_job_before_starting(self, current_status: str) -> bool:
validate_job_status(current_status, JobStatus.job_statuses())
return current_status in [
JobStatus.SUBMITTED,
JobStatus.PENDING,
JobStatus.RUNNABLE,
]
@classmethod
def status_transitions(self) -> list[tuple[Optional[str], str]]:
return [
(JobStatus.SUBMITTED.value, JobStatus.PENDING.value),
(JobStatus.PENDING.value, JobStatus.RUNNABLE.value),
(JobStatus.RUNNABLE.value, JobStatus.STARTING),
(JobStatus.STARTING.value, JobStatus.RUNNING.value),
]
|
