simono41/compose-viz
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge branch 'main' of https://github.com/compose-viz/compose-viz into dev

Browse source
This commit is contained in:
Xyphuz 2023-01-10 22:07:10 +08:00
commit 2a815e1fba
3 changed files with 39 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
README.md
View file

@ -119,6 +119,7 @@ Check out the result [here](https://github.com/compose-viz/compose-viz/blob/main
| ----------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `-o, --output-filename` | Output filename for the generated visualization file. [default: compose-viz] |
| `-m, --format` | Output format for the generated visualization file. See [supported formats](https://github.com/compose-viz/compose-viz/blob/main/compose_viz/models/viz_formats.py). [default: png] |
| `-r, --root-service` | Root of the service tree (convenient for large compose yamls) |
| `-v, --version` | Show the version of compose-viz. |
| `--help` | Show help and exit. |

8
compose_viz/cli.py
View file

@ -36,6 +36,12 @@ def compose_viz(
"-m",
help="Output format for the generated visualization file.",
),
root_service: str = typer.Option(
None,
"--root-service",
"-r",
help="Root of the service tree (convenient for large compose yamls)",
),
_: Optional[bool] = typer.Option(
None,
"--version",
@ -46,7 +52,7 @@ def compose_viz(
),
) -> None:
parser = Parser()
compose = parser.parse(input_path)
compose = parser.parse(input_path, root_service=root_service)
if compose:
typer.echo(f"Successfully parsed {input_path}")

37
compose_viz/parser.py
View file

@ -1,5 +1,5 @@
import re
from typing import List, Optional
from typing import Any, Dict, List, Optional, Union
from pydantic import ValidationError
@ -15,7 +15,27 @@ class Parser:
def __init__(self):
pass
def parse(self, file_path: str) -> Compose:
@staticmethod
def _unwrap_depends_on(data_depends_on: Union[spec.ListOfStrings, Dict[Any, spec.DependsOn], None]) -> List[str]:
service_depends_on = []
if type(data_depends_on) is spec.ListOfStrings:
service_depends_on = data_depends_on.__root__
elif type(data_depends_on) is dict:
for depends_on in data_depends_on.keys():
service_depends_on.append(str(depends_on))
return service_depends_on
@staticmethod
def compile_dependencies(service_name: str, compose_data: spec.ComposeSpecification) -> List[str]:
assert compose_data.services
dependencies = []
for dependency in Parser._unwrap_depends_on(compose_data.services[service_name].depends_on):
if dependency:
dependencies.append(dependency)
dependencies.extend(Parser.compile_dependencies(dependency, compose_data))
return dependencies
def parse(self, file_path: str, root_service: Optional[str] = None) -> Compose:
compose_data: spec.ComposeSpecification
try:
@ -27,8 +47,15 @@ class Parser:
assert compose_data.services is not None, "No services found, aborting."
root_dependencies: List[str] = list()
if root_service:
root_dependencies = Parser.compile_dependencies(root_service, compose_data)
root_dependencies.append(root_service)
for service_name, service_data in compose_data.services.items():
service_name = str(service_name)
if root_service and service_name not in root_dependencies:
continue
service_image: Optional[str] = None
if service_data.build is not None:
@ -132,11 +159,7 @@ class Parser:
service_depends_on: List[str] = []
if service_data.depends_on is not None:
if type(service_data.depends_on) is spec.ListOfStrings:
service_depends_on = service_data.depends_on.__root__
elif type(service_data.depends_on) is dict:
for depends_on in service_data.depends_on.keys():
service_depends_on.append(str(depends_on))
service_depends_on = Parser._unwrap_depends_on(service_data.depends_on)
service_volumes: List[Volume] = []
if service_data.volumes is not None: