-
Notifications
You must be signed in to change notification settings - Fork 0
/
services.py
192 lines (153 loc) · 5.88 KB
/
services.py
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
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
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
"""Define all the orchestration functionality required by the program to work.
Classes and functions that connect the different domain model objects with the adapters
and handlers to achieve the program's purpose.
"""
import logging
import time
from typing import Dict, Optional
from .adapters import Drone
from .adapters.aws import AWS, AutoscalerInfo
from .adapters.drone import DronePromoteError
from .config import Config, ConfigError
log = logging.getLogger(__name__)
def wait(
drone: Drone, project_pipeline: str, build_number: Optional[int] = None
) -> bool:
"""Wait for the pipeline build to finish.
Args:
project_pipeline: Drone pipeline identifier.
In the format of `repo_owner/repo_name`.
build_number: Number of drone build.
Returns:
True: When job has finished.
Raises:
DroneBuildError: if the job doesn't exist
DroneAPIError: if the API returns a job with a "number" that is not an int.
"""
if build_number is None:
last_build = drone.last_build_info(project_pipeline)
if last_build.finished != 0:
log.info("There are no active jobs")
return True
build_number = last_build.number
first_time = True
while True:
build = drone.build_info(project_pipeline, build_number)
if build.finished == 0:
if first_time:
log.info(
f"Waiting for job #{build.number} started by "
f"a {build.event} event by {build.trigger}."
)
first_time = False
time.sleep(1)
continue
log.info(f"Job #{build.number} has finished with status {build.status}")
return True
def _check_project(config: Config, project_id: str) -> None:
"""Check if the project_id exists.
Raises:
ConfigError: if project doesn't exist
"""
if project_id not in config["projects"].keys():
raise ConfigError(f"The project {project_id} does not exist")
def get_active_project(config: Config) -> str:
"""Return the active project id.
Returns:
project_id: Active project id
Raises:
ConfigError: If there are no active projects, no configured projects or
the active project doesn't exist.
"""
try:
if config["active_project"] is None:
raise KeyError("There are no active projects.")
_check_project(config, config["active_project"])
return config["active_project"]
except KeyError as error:
try:
if len(config["projects"].keys()) == 1:
return [key for key, value in config["projects"].items()][0]
raise ConfigError(
"There are more than one project configured but none "
"is marked as active. Please use drode set command to "
"define one."
) from error
except (KeyError, AttributeError):
raise ConfigError("There are no projects configured.") from error
def set_active_project(config: Config, project_id: str) -> None:
"""Set the active project.
Raises:
ConfigError: If the project to activate doesn't exist.
"""
_check_project(config, project_id)
config["active_project"] = project_id
config.save()
log.info(f"The project {project_id} is now active")
def ask(question: str) -> bool:
"""Prompt the user to answer yes or no to a question.
Returns:
answer: User's answer
"""
answer = input(question)
if answer in ["yes", "y"]:
return True
return False
def promote(
drone: Drone,
project_pipeline: str,
environment: str,
build_number: Optional[int] = None,
) -> Optional[int]:
"""Promote build_number or commit id to the desired environment.
Args:
drone: Drone adapter.
build_number: Number of drone build or commit id.
project_pipeline: Drone pipeline identifier.
In the format of `repo_owner/repo_name`.
environment: Environment one of ['production', 'staging']
Returns:
build_number: Job that promotes the desired build number.
Raises:
DroneAPIError: if the returned build information contains an after that is not
a string.
"""
if build_number is None:
build = drone.last_success_build_info(project_pipeline)
else:
build = drone.build_info(project_pipeline, build_number)
if build.status != "success":
raise DronePromoteError(
f"You can't promote job #{build.number} to {environment} "
f"as it's status is {build.status}"
)
log.info(
f"You're about to promote job #{build.number} "
f"of the pipeline {project_pipeline} to {environment}\n\n"
f" With commit {build.after[:8]}: {build.message}"
)
if ask("Are you sure? [y/N]: "):
return drone.promote(project_pipeline, build.number, environment)
return None
ProjectStatus = Dict[str, AutoscalerInfo]
def project_status(config: Config, aws: AWS) -> ProjectStatus:
"""Fetch the status of the autoscaling groups of the active project.
Raises:
ConfigError: If there are no active projects, no configured projects or
the active project doesn't exist.
"""
project: ProjectStatus = {}
active_project = get_active_project(config)
for environment in ["Production", "Staging"]:
try:
autoscaler_name = config.get(
f"projects.{active_project}."
f"aws.autoscaling_groups.{environment.lower()}"
)
if not isinstance(autoscaler_name, str):
raise ConfigError("The autoscaler name is not a string")
autoscaler_info = aws.get_autoscaling_group(autoscaler_name)
except ConfigError:
autoscaler_info = AutoscalerInfo(instances=[])
project[environment] = autoscaler_info
return project