Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .dockerignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
.github/
.pytest_cache/
.ruff_cache/
.gitignore
__pycache__/
venv/
4 changes: 2 additions & 2 deletions Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ COPY cli/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir pytest

COPY cli /app/cli
COPY . /app

ENTRYPOINT ["python", "cli/app.py"]
ENTRYPOINT ["python", "-m", "cli.app"]
CMD [""]
Empty file added cli/__init__.py
Empty file.
26 changes: 26 additions & 0 deletions cli/app.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,40 @@
import os
import subprocess
import typer

from pathlib import Path
from rich import print
from typing import Annotated

from config import config
from .merge import merge

try:
config_data = config.load_config_file(Path("config/config.toml"))
except Exception:
config_data = {}
print(f"Check that a config.toml file is populated here: '{Path.home()}'")
print("Common Problems: an API key is not set, or is invalid.")

app = typer.Typer(
help="Surge - A DevOps CLI Tool For System Monitoring and Production Reliability"
)

# Merges app.command() decorator w/ transposed merge() decorator
cmd = app.command


def app_command_with_merge(*args, **kwargs):
decorator = cmd(*args, **kwargs)

def wrapper(func):
return decorator(merge()(func))

return wrapper


app.command = app_command_with_merge


def run_cmd(cmd: str) -> str:
"""
Expand Down
82 changes: 82 additions & 0 deletions cli/merge.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
import inspect
from functools import wraps


def merge(section: str | None = None):
"""
Decorator to merge the func call arguments/options of Typer with config defaults provided via TOML tables.
The precedence is as follows:
- Explicit CLI Arguments: (i.e. surge <cmd> -<option-flag> --<option-name>)

- Configuration Data for a given section (
found in TOML table under table, i.e.:
[table-name]
<option> = <value>

# Provided as a KWArg for Python in config.py
)

- Typer function default signature (
i.e.
def name(option: typer.Option('-x', '--flag') = <value>, ...):
)
"""

def decorator(func):
orig_sig = inspect.signature(func)
declared_defaults = {
name: param.default
for name, param in orig_sig.parameters.items()
if param.default is not inspect._empty
}

# Modded signature with defaults set to None for Typer to see as optional; leaves *args, **kwargs, and positional-only params default
new_params = []
for param in orig_sig.parameters.values():
if param.kind in (
inspect.Parameter.VAR_POSITIONAL,
inspect.Parameter.VAR_KEYWORD,
inspect.Parameter.POSITIONAL_ONLY,
):
new_params.append(param)
else:
new_params.append(param.replace(default=None))
new_sig = orig_sig.replace(parameters=new_params)

# Actual wrapper lets go
@wraps(func)
def wrapper(*args, **kwargs):
config_data = func.__globals__.get("config_data", {})

section_key = section or func.__name__
config_section = (
config_data.get(section_key, {})
if isinstance(config_data, dict)
else {}
)

bound = orig_sig.bind_partial(*args, **kwargs)
final_args = {}

for name in orig_sig.parameters:
val = bound.arguments.get(name, None)

if val is not None: # Explicit cli option
final_args[name] = val
continue

if name in config_section: # Config option
final_args[name] = config_section[name]
continue

if name in declared_defaults: # Default fallback
final_args[name] = declared_defaults[name]
else:
final_args[name] = None

return func(**final_args)

wrapper.__signature__ = new_sig
return wrapper

return decorator
1 change: 1 addition & 0 deletions cli/requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ shellingham
sniffio
SQLAlchemy
tenacity
tomli-w
typer
typing-inspection
typing_extensions
Expand Down
Empty file added config/__init__.py
Empty file.
57 changes: 57 additions & 0 deletions config/config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
import tomllib
import tomli_w
from pathlib import Path

PATH = Path("config/config.toml")

DEFAULT_DATA = {
"console": {"force_color": True, "theme": "default"},
"monitor": {
"interval": 5,
"load": True,
"cpu": True,
"ram": True,
"disk": True,
"io": False,
"verbose": False,
},
"network": {"requests": 5, "dtype": "A", "sockets": False, "no_trace": False},
"ai": {"format": "hybrid", "verbosity": "normal", "auto_fix": False},
}


def create_config_file(path: Path) -> None:
try:
with open(path, "wb") as config:
tomli_w.dump(DEFAULT_DATA, config)
print(f"Created config file at {Path.home()}")
except Exception as e:
print(f"Could not create config.toml file: {e}")


def load_config_file(path: Path) -> dict:
if not path.exists():
print("Config file does not exist, creating new config file...")
create_config_file(path)
return DEFAULT_DATA

with open(path, "rb") as config:
data = tomllib.load(config)

if not data:
print("Config file is empty, creating new config file...")
create_config_file(path)
return DEFAULT_DATA

return data


if __name__ == "__main__":
"""
Run this file for a default config.toml file
"""
try:
load_config_file(PATH)
print(f"Loaded config file at {PATH}")
except Exception:
create_config_file()
23 changes: 23 additions & 0 deletions config/config.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
[console]
force_color = true
theme = "default"

[monitor]
interval = 5
load = true
cpu = true
ram = true
disk = true
io = false
verbose = false

[network]
requests = 5
dtype = "A"
sockets = false
no_trace = false

[ai]
format = "hybrid"
verbosity = "normal"
auto_fix = false