-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
52 lines (40 loc) · 1.61 KB
/
Copy pathmain.py
File metadata and controls
52 lines (40 loc) · 1.61 KB
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
import os
import sys
from multiprocessing import Process, freeze_support
from typing import TYPE_CHECKING
import logging
from models.configuration import load_configuration
from models.models_file import AppConfiguration
from server import Server
from utils.logger import create_logger
if len(sys.argv) == 1: sys.argv.append(os.getcwd())
# this line assures execution with cwd as argument when running main.py as script.
if TYPE_CHECKING:
from typing import List
def get_servers(config) -> 'List[Server]':
environments = load_configuration(config.mockoon_file)
servers = [Server.factory(config.host, environment, debug=config.flask_debug, logging_level=config.logging_level)
for environment in environments]
return servers
def main():
config = AppConfiguration.fromFile("config.yaml")
logger = create_logger(level=config.logging_level)
try:
servers = get_servers(config)
except Exception as e:
logger.error("Error in get_servers(): %s", e)
raise SystemExit
processes = [Process(target=server.run, name=server.name) for server in servers]
yield processes
[proc.start() for proc in processes]
running_processes = [process for process in processes if process.is_alive()]
logger.info("Started %s servers out of %s requested: %s", len(running_processes), len(servers),
" ".join([process.name for process in running_processes]))
[proc.join() for proc in processes]
yield None
if __name__ == '__main__':
freeze_support()
sequence = main()
processes = next(sequence)
len(processes)
start = next(sequence)