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
|
import argparse
import os
import signal
import sys
from typing import Any, Optional
from werkzeug.serving import run_simple
# Expose ThreadedMotoServer as a public symbol according to PEP-561
from moto.moto_server.threaded_moto_server import (
ThreadedMotoServer as ThreadedMotoServer,
)
from moto.moto_server.werkzeug_app import (
DomainDispatcherApplication,
create_backend_app,
)
def signal_handler(signum: Any, frame: Any) -> None:
sys.exit(0)
def main(argv: Optional[list[str]] = None) -> None:
argv = argv or sys.argv[1:]
parser = argparse.ArgumentParser()
parser.add_argument(
"-H", "--host", type=str, help="Which host to bind", default="127.0.0.1"
)
parser.add_argument(
"-p",
"--port",
type=int,
help="Port number to use for connection",
default=int(os.environ.get("MOTO_PORT", 5000)),
)
parser.add_argument(
"-r",
"--reload",
action="store_true",
help="Reload server on a file change",
default=False,
)
parser.add_argument(
"-s",
"--ssl",
action="store_true",
help="Enable SSL encrypted connection with auto-generated certificate (use https://... URL)",
default=False,
)
parser.add_argument(
"-c", "--ssl-cert", type=str, help="Path to SSL certificate", default=None
)
parser.add_argument(
"-k", "--ssl-key", type=str, help="Path to SSL private key", default=None
)
args = parser.parse_args(argv)
if "MOTO_PORT" not in os.environ:
os.environ["MOTO_PORT"] = f"{args.port}"
try:
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
except Exception:
pass # ignore "ValueError: signal only works in main thread"
# Wrap the main application
main_app = DomainDispatcherApplication(create_backend_app)
main_app.debug = True # type: ignore
ssl_context: Any = None
if args.ssl_key and args.ssl_cert:
ssl_context = (args.ssl_cert, args.ssl_key)
elif args.ssl:
ssl_context = "adhoc"
run_simple(
args.host,
args.port,
main_app,
threaded=True,
use_reloader=args.reload,
ssl_context=ssl_context,
)
if __name__ == "__main__":
main()
|
