121 lines
3.2 KiB
Python
121 lines
3.2 KiB
Python
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
from click.termui import _ansi_colors as ansi_colors # type: ignore
|
|
from dotenv import dotenv_values
|
|
|
|
from catfish.utils.processes import get_running_process_for
|
|
|
|
PROCFILE_LOCATIONS = ["Procfile", "etc/environments/development/Procfile"]
|
|
ENVFILE_LOCATIONS = [".env", "etc/environments/development/env"]
|
|
PROCFILE_LINE = re.compile(r"^([A-Za-z0-9_]+):\s*(.+)$")
|
|
|
|
|
|
class DuplicateProcessException(Exception):
|
|
pass
|
|
|
|
|
|
def parse_procfile_processes(project, procfile_lines):
|
|
seen_names = []
|
|
for line in procfile_lines:
|
|
m = PROCFILE_LINE.match(line)
|
|
if m:
|
|
if m.group(1) in seen_names:
|
|
raise DuplicateProcessException(m.group(1))
|
|
seen_names.append(m.group(1))
|
|
yield Process(project, m.group(1), m.group(2))
|
|
|
|
|
|
@dataclass
|
|
class Project:
|
|
root: Path
|
|
processes: List["Process"] = field(default_factory=list)
|
|
env: Dict[str, str] = field(default_factory=dict)
|
|
|
|
def __post_init__(self):
|
|
self.root = Path(self.root)
|
|
assert self.exists()
|
|
self.processes = self.read_processes()
|
|
self.read_envfiles()
|
|
|
|
def read_processes(self):
|
|
for location in PROCFILE_LOCATIONS:
|
|
procfile_path = self.root.joinpath(location)
|
|
if procfile_path.exists():
|
|
with procfile_path.open() as f:
|
|
return list(parse_procfile_processes(self, f.readlines()))
|
|
return []
|
|
|
|
def read_envfiles(self):
|
|
for location in ENVFILE_LOCATIONS:
|
|
envfile_path = self.root.joinpath(location)
|
|
if envfile_path.exists():
|
|
self.env.update(dotenv_values(envfile_path))
|
|
|
|
def exists(self):
|
|
return self.root.exists() and self.root.is_dir()
|
|
|
|
@property
|
|
def name(self):
|
|
return self.root.name
|
|
|
|
def get_process(self, name):
|
|
try:
|
|
return next(proc for proc in self.processes if proc.name == name)
|
|
except StopIteration:
|
|
return None
|
|
|
|
def get_extra_path(self):
|
|
return [self.root.joinpath("node_modules/.bin"), self.root.joinpath("env/bin")]
|
|
|
|
def get_environment(self):
|
|
return {
|
|
**self.env,
|
|
"PYTHONUNBUFFERED": "1",
|
|
"PATH": "{}:{}".format(
|
|
":".join(map(str, self.get_extra_path())), os.environ["PATH"]
|
|
),
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Process:
|
|
project: Project
|
|
name: str
|
|
command: str
|
|
|
|
@property
|
|
def ident(self):
|
|
return self.project.name + ":" + self.name
|
|
|
|
@property
|
|
def logs_socket(self):
|
|
return f"{self.project.name}-{self.name}.sock"
|
|
|
|
def get_running_process(self):
|
|
return get_running_process_for(self)
|
|
|
|
@property
|
|
def is_running(self):
|
|
return self.get_running_process() is not None
|
|
|
|
@property
|
|
def pid(self):
|
|
proc = self.get_running_process()
|
|
if proc:
|
|
return proc.pid
|
|
|
|
@property
|
|
def port(self):
|
|
proc = self.get_running_process()
|
|
if proc:
|
|
return proc.pid
|
|
|
|
@property
|
|
def color(self):
|
|
proc_index = self.project.processes.index(self)
|
|
return list(ansi_colors.keys())[1:][proc_index % len(ansi_colors)]
|