from collections import OrderedDict
from collections import defaultdict
from collections import namedtuple
import shlex
import os
import re
PROCFILE_LINE = re.compile(r'^([A-Za-z0-9_-]+):\s*(.+)$')
[docs]class Env(object):
def __init__(self, config):
self._c = config
@property
def port(self):
try:
return int(self._c['port'])
except ValueError:
raise ValueError(f"invalid value for port: '{self._c['port']}'")
@property
def procfile(self):
return os.path.join(self._c['app_root'], self._c['procfile'])
[docs] def load_procfile(self):
with open(self.procfile) as f:
content = f.read()
return parse_procfile(content)
[docs]class Procfile(object):
"""A data structure representing a Procfile"""
def __init__(self):
self.processes = OrderedDict()
[docs] def add_process(self, name, command):
assert name not in self.processes, \
"process names must be unique within a Procfile"
self.processes[name] = command
[docs]def parse_procfile(contents):
p = Procfile()
for line in contents.splitlines():
m = PROCFILE_LINE.match(line)
if m:
p.add_process(m.group(1), m.group(2))
return p
[docs]def parse(content):
"""
Parse the content of a .env file (a line-delimited KEY=value format) into a
dictionary mapping keys to values.
"""
values = {}
for line in content.splitlines():
lexer = shlex.shlex(line, posix=True)
tokens = list(lexer)
# parses the assignment statement
if len(tokens) < 3:
continue
name, op = tokens[:2]
value = ''.join(tokens[2:])
if op != '=':
continue
if not re.match(r'[A-Za-z_][A-Za-z_0-9]*', name):
continue
value = value.replace(r'\n', '\n')
value = value.replace(r'\t', '\t')
values[name] = value
return values
ProcessParams = namedtuple("ProcessParams", "name cmd quiet env")
[docs]def expand_processes(processes, concurrency=None, env=None, quiet=None, port=None):
"""
Get a list of the processes that need to be started given the specified
list of process types, concurrency, environment, quietness, and base port
number.
Returns a list of ProcessParams objects, which have `name`, `cmd`, `env`,
and `quiet` attributes, corresponding to the parameters to the constructor
of `honcho.process.Process`.
"""
if env is not None and env.get("PORT") is not None:
port = int(env.get("PORT"))
if quiet is None:
quiet = []
con = defaultdict(lambda: 1)
if concurrency is not None:
con.update(concurrency)
out = []
for name, cmd in processes.items():
for i in range(con[name]):
n = "{0}.{1}".format(name, i + 1)
c = cmd
q = name in quiet
e = {'HONCHO_PROCESS_NAME': n}
if env is not None:
e.update(env)
if port is not None:
e['PORT'] = str(port + i)
params = ProcessParams(n, c, q, e)
out.append(params)
if port is not None:
port += 100
return out