-
-
Notifications
You must be signed in to change notification settings - Fork 100
/
commands.py
213 lines (191 loc) · 6.34 KB
/
commands.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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
import click
import uuid
import boto3
import os
from mangum.platforms.aws.helpers import (
get_default_resource_name,
get_default_region_name,
get_log_events,
get_config,
)
from mangum.platforms.aws.config import AWSConfig
class AWSGroup(click.Group):
def get_command(self, ctx, cmd_name):
rv = click.Group.get_command(self, ctx, cmd_name)
if rv is not None:
return rv
matches = [x for x in self.list_commands(ctx) if x.startswith(cmd_name)]
if not matches:
return None
elif len(matches) == 1:
return click.Group.get_command(self, ctx, matches[0])
ctx.fail("Too many matches: %s" % ", ".join(sorted(matches)))
@click.command(cls=AWSGroup)
def aws() -> None:
pass
@aws.command()
def init() -> None:
click.echo(f"Welcome to Mangum, serverless ASGI!")
project_name = click.prompt("What is the name of the project?", type=str)
click.echo(f"{project_name}, great!")
project_dir_name = click.prompt(
"What is the name of the directory that contains the project code?", type=str
)
config_dir = os.getcwd()
project_dir = os.path.join(config_dir, project_dir_name)
if not os.path.isdir(project_dir):
raise click.ClickException(
f"Directory not found at {project_dir}! "
"(Hint: The project folder must be in current working directory.)"
)
description = click.prompt(
"Enter a brief description of your project.",
type=str,
default="ASGI application",
)
resource_name = click.prompt(
"What should be the prefix used for naming resources?",
type=str,
default=get_default_resource_name(project_name),
)
url_root = click.prompt("What should be the root URL path?", type=str, default="/")
runtime_version = click.prompt(
"What version of Python are you using?", type=str, default="3.7"
)
timeout = click.prompt(
"What should the timeout be (in seconds, max=300)?", type=int, default=300
)
default_region_name = get_default_region_name()
region_name = click.prompt(
"What region should be used?", default=default_region_name
)
s3_bucket_name = click.prompt(
"An S3 bucket is required. \n\nEnter the name of an existing bucket or "
f"one will be generated.",
type=str,
default="",
)
generate_s3 = s3_bucket_name == ""
if generate_s3:
s3_bucket_name = f"{resource_name.lower()}-{uuid.uuid4()}"
s3 = boto3.resource("s3")
s3.create_bucket(
Bucket=s3_bucket_name,
CreateBucketConfiguration={"LocationConstraint": region_name},
)
config = AWSConfig(
project_name=project_name,
description=description,
s3_bucket_name=s3_bucket_name,
resource_name=resource_name,
url_root=url_root,
runtime_version=runtime_version,
region_name=region_name,
timeout=timeout,
stack_name=resource_name.lower(),
)
click.echo("Creating your local project...")
config.init()
click.echo(
"Your app configuration has been generated!\n"
"Modify the 'template.yaml' or 'requirements.txt' if you need to, otherwise "
"run 'mangum aws build' to prepare the local project for packaging."
)
@aws.command()
def build() -> None:
"""
Install the packages listed in the requirements file and copy the application files
to the 'build/' directory to prepare for packaging.
"""
config, error = get_config()
if error is not None:
click.echo(error)
else:
click.echo("Building application and installing requirements...")
config.build()
click.echo("Build complete!")
@aws.command()
def update() -> None:
"""
Update the application files only.
"""
config, error = get_config()
if error is not None:
click.echo(error)
else:
click.echo("Updating the application build.")
config.build(update=True)
click.echo("Update complete!")
@aws.command()
def package() -> None:
config, error = get_config()
if error is not None:
click.echo(error)
else:
click.echo("Packaging your application...")
packaged = config.cli_package()
if not packaged:
click.echo("There was an error...")
else:
click.echo(
"Successfully packaged. Run 'mangum aws deploy' to deploy it now."
)
@aws.command()
def deploy() -> None:
config, error = get_config()
if error is not None:
click.echo(error)
else:
click.echo("Deploying your application! This may take some time...")
deployed = config.cli_deploy()
if not deployed:
click.echo("There was an error...")
else:
endpoints = config.cli_describe()
click.echo(
f"Deployment successful! API endpoints available at:\n\n{endpoints}"
)
@aws.command()
def describe() -> None:
config, error = get_config()
if error is not None:
click.echo(error)
else:
endpoints = config.cli_describe()
if not endpoints:
click.echo("Error! Could not retrieve endpoints.")
else:
click.echo(f"API endpoints available at:\n\n{endpoints}")
@aws.command()
def validate() -> None:
config, error = get_config()
if error is not None:
click.echo(error)
else:
error = config.validate()
if not error:
click.echo("Template file validated successfully!")
else:
click.echo(error)
@aws.command()
def tail() -> None:
config, error = get_config()
if error is not None:
click.echo(error)
else:
# Display the CloudWatch logs for the last 10 minutes.
# TODO: Make this configurable.
log_events = get_log_events(
f"/aws/lambda/{config.resource_name}Function", minutes=10
)
log_output = []
for log in log_events:
message = log["message"].rstrip()
if not any(
i in message
for i in ("START RequestId", "REPORT RequestId", "END RequestId")
):
timestamp = log["timestamp"]
s = f"[{timestamp}] {message}"
log_output.append(s)
click.echo("\n".join(log_output))