forked from opsmill/infrahub-demo-dc-fabric
-
Notifications
You must be signed in to change notification settings - Fork 3
/
tasks.py
115 lines (82 loc) · 2.9 KB
/
tasks.py
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
import os
from pathlib import Path
from invoke import task, Context # type: ignore
MAIN_DIRECTORY_PATH = Path(__file__).parent
DATA_GENERATORS = [
"create_basic.py",
"create_location.py",
"create_topology.py",
"create_security_nodes.py",
]
# If no version is indicated, we will take the latest
VERSION = os.getenv("INFRAHUB_VERSION", None)
COMPOSE_COMMAND = f"curl https://infrahub.opsmill.io/{VERSION if VERSION else ''} | docker compose -f -"
COMPOSE_COMMAND_LOCAL = "docker compose"
def has_local_docker_file() -> bool:
file_path = Path(MAIN_DIRECTORY_PATH, "docker-compose.yml")
return file_path.is_file()
def get_docker_command() -> str:
if has_local_docker_file():
return COMPOSE_COMMAND_LOCAL
return COMPOSE_COMMAND
@task
def start(context: Context) -> None:
with context.cd(MAIN_DIRECTORY_PATH):
context.run(f"{get_docker_command()} up -d")
@task
def load_schema(context: Context, schema: Path = Path("./models/*.yml")) -> None:
context.run(f"infrahubctl schema load {schema} --wait 30")
@task
def load_data(context: Context) -> None:
with context.cd(MAIN_DIRECTORY_PATH):
for generator in DATA_GENERATORS:
context.run(f"infrahubctl run bootstrap/{generator}")
@task
def destroy(context: Context) -> None:
with context.cd(MAIN_DIRECTORY_PATH):
context.run(f"{get_docker_command()} down -v")
@task
def stop(context: Context) -> None:
with context.cd(MAIN_DIRECTORY_PATH):
context.run(f"{get_docker_command()} down")
@task
def restart(context: Context, component: str = "") -> None:
with context.cd(MAIN_DIRECTORY_PATH):
if component:
context.run(f"{get_docker_command()} restart {component}")
return
context.run(f"{get_docker_command()} restart")
@task
def format(context: Context) -> None:
"""Run RUFF to format all Python files."""
exec_cmds = ["ruff format .", "ruff check . --fix"]
with context.cd(MAIN_DIRECTORY_PATH):
for cmd in exec_cmds:
context.run(cmd)
@task
def lint_yaml(context: Context) -> None:
"""Run Linter to check all Python files."""
print(" - Check code with yamllint")
exec_cmd = "yamllint ."
with context.cd(MAIN_DIRECTORY_PATH):
context.run(exec_cmd)
@task
def lint_mypy(context: Context) -> None:
"""Run Linter to check all Python files."""
print(" - Check code with mypy")
exec_cmd = "mypy --show-error-codes infrahub_sdk"
with context.cd(MAIN_DIRECTORY_PATH):
context.run(exec_cmd)
@task
def lint_ruff(context: Context) -> None:
"""Run Linter to check all Python files."""
print(" - Check code with ruff")
exec_cmd = "ruff check ."
with context.cd(MAIN_DIRECTORY_PATH):
context.run(exec_cmd)
@task(name="lint")
def lint_all(context: Context) -> None:
"""Run all linters."""
lint_yaml(context)
lint_ruff(context)
lint_mypy(context)