Template syntax error: {}, line {}
" - ).format(e.message, e.lineno) - base.error = error_message - print(base) - return HTMLResponse(content=error_message) - except Exception as e: - LOGGER.exception("Error rendering template") - trace = markupsafe.escape(traceback.format_exc()) - error_message = markupsafe.Markup( - '' - f"Unexpected error: {type(e).__name__}: {e}" - '
'
- f"object={object!r}\nmodel={self.model!r}"
- f"\n\n{trace}"
- ""
- )
- return HTMLResponse(content=error_message)
-
- def configure_routes(self): # noqa: C901
- self.app.mount(
- f"{ROUTE_PREFIX}/assets",
- StaticFiles(
- directory=PATH_TO_FRONTEND.joinpath("assets"), html=True
- ),
- )
- self.app.mount(
- f"{ROUTE_PREFIX}/static",
- StaticFiles(
- directory=PATH_TO_FRONTEND.joinpath("static"), html=True
- ),
- )
-
- @self.router.get("/api/views")
- def read_templates():
- self.templates_index = self.templates_loader.index_path(
- self.templates_path
- )
- return self.templates_index.as_dict
-
- @self.router.get("/api/objects/{uuid}")
- def read_object(uuid: str):
- obj = self.model.by_uuid(uuid)
- details = tl.simple_object(obj)
- return {
- "idx": details["idx"],
- "name": details["name"],
- "type": obj.xtype,
- }
-
- @self.router.get("/api/views/{template_name}")
- def read_template(template_name: str):
- template_name = urlparse.unquote(template_name)
- if (
- self.templates_index is None
- or template_name not in self.templates_index.flat
- ):
- return {
- "error": (
- f"Template {template_name} not found"
- " or templates index not initialized"
- )
- }
- base = self.templates_index.flat[template_name]
- base.compute_instance_list(self.model)
- return base
-
- @self.router.get("/api/views/{template_name}/{object_id}")
- def render_template(template_name: str, object_id: str):
- content = None
- object = None
- try:
- template_name = urlparse.unquote(template_name)
- if (
- self.templates_index is None
- or template_name not in self.templates_index.flat
- ):
- return {"error": f"Template {template_name} not found"}
- base = self.templates_index.flat[template_name]
- template_filename = base.template
- # load the template file from the templates folder
- content = (self.templates_path / template_filename).read_text(
- encoding="utf8"
- )
- except Exception as e:
- error_message = markupsafe.Markup(
- "Template not found: {}
" - ).format(str(e)) - return HTMLResponse(content=error_message) - if object_id == "render": - object = None - else: - try: - object = self.model.by_uuid(object_id) - except Exception as e: - error_message = markupsafe.Markup( - "" - "Requested object not found: {}
" - ).format(str(e)) - return HTMLResponse(content=error_message) - return self.render_instance_page(content, base, object) - - @self.router.get("/api/model-info") - def model_info(): - info = self.model.info - resinfo = info.resources["\x00"] - return { - "title": info.title, - "revision": resinfo.revision, - "hash": resinfo.rev_hash, - "capella_version": info.capella_version, - "branch": resinfo.branch, - "badge": self.model.description_badge, - } - - @self.app.get("/metrics") - def metrics(): - idle_time_minutes = (time.time() - self.last_interaction) / 60 - self.idle_time_gauge.set(idle_time_minutes) - return fastapi.Response( - content=prometheus_client.generate_latest(), - media_type="text/plain", - ) - - @self.router.get("/api/metadata") - async def version(): - return {"version": self.app.version} - - @self.router.post("/api/compare") - async def post_compare(commit_range: CommitRange): - try: - self.diff = model_diff.get_diff_data( - self.model, commit_range.head, commit_range.prev - ) - self.diff["lookup"] = create_diff_lookup(self.diff["objects"]) - if self.diff["lookup"]: - return {"success": True} - return {"success": False, "error": "No model changes to show"} - except Exception as e: - LOGGER.exception("Failed to compare versions") - return {"success": False, "error": str(e)} - - @self.router.post("/api/object-diff") - async def post_object_diff(object_id: ObjectDiffID): - if object_id.uuid not in self.diff["lookup"]: - raise HTTPException(status_code=404, detail="Object not found") - - self.object_diff = self.diff["lookup"][object_id.uuid] - return {"success": True} - - @self.router.get("/api/commits") - async def get_commits(): - try: - return model_diff.populate_commits(self.model) - except Exception as e: - return {"error": str(e)} - - @self.router.get("/api/diff") - async def get_diff(): - if self.diff: - return self.diff - return { - "error": "No data available. Please compare two commits first." - } - - @self.router.get("/{rest_of_path:path}") - async def catch_all(request: Request, rest_of_path: str): - del rest_of_path - return self.app.state.templates.TemplateResponse( - "index.html", {"request": request} - ) - - -def index_template(template, templates, templates_grouped, filename=None): - idx = filename if filename else template["idx"] - record = {"idx": idx, **template} - if "category" in template: - category = template["category"] - if category not in templates_grouped: - templates_grouped[category] = [] - templates_grouped[category].append(record) - else: - templates_grouped["other"].append(record) - templates[idx] = template - - -def index_templates( - path: pathlib.Path, -) -> tuple[dict[str, t.Any], dict[str, t.Any]]: - templates_grouped: dict[str, t.Any] = {"other": []} - templates: dict[str, t.Any] = {} - for template_file in path.glob("**/*.yaml"): - template = yaml.safe_load(template_file.read_text(encoding="utf8")) - if "templates" in template: - for template_def in template["templates"]: - index_template(template_def, templates, templates_grouped) - else: - idx = urlparse.quote(template_file.name.replace(".yaml", "")) - index_template( - template, templates, templates_grouped, filename=idx - ) - return templates_grouped, templates - - -def create_diff_lookup(data, lookup=None): - if lookup is None: - lookup = {} - try: - if isinstance(data, dict): - for _, obj in data.items(): - if "uuid" in obj: - lookup[obj["uuid"]] = { - "uuid": obj["uuid"], - "display_name": obj["display_name"], - "change": obj["change"], - "attributes": obj["attributes"], - } - if children := obj.get("children"): - create_diff_lookup(children, lookup) - except Exception: - LOGGER.exception("Cannot create diff lookup") - return lookup diff --git a/capella_model_explorer/backend/main.py b/capella_model_explorer/backend/main.py new file mode 100644 index 0000000..9cba147 --- /dev/null +++ b/capella_model_explorer/backend/main.py @@ -0,0 +1,325 @@ +# Copyright DB InfraGO AG and contributors +# SPDX-License-Identifier: Apache-2.0 + +import logging +import os +import pathlib +import time +import traceback +import typing as t +import urllib.parse + +import capellambse +import capellambse.model as m +import fastapi +import fastapi.middleware.cors +import fastapi.responses +import fastapi.staticfiles +import fastapi.templating +import jinja2 +import markupsafe +import prometheus_client +import pydantic + +import capella_model_explorer +from capella_model_explorer.backend import model_diff, state +from capella_model_explorer.backend import templates as tl + +FRONTEND_DIR = pathlib.Path("./frontend/dist") +ROUTE_PREFIX = os.getenv("ROUTE_PREFIX", "") +logger = logging.getLogger(__name__) + + +class CommitRange(pydantic.BaseModel): + head: str + prev: str + + +class ObjectDiffID(pydantic.BaseModel): + uuid: str + + +def _create_diff_lookup(data, lookup=None): + if lookup is None: + lookup = {} + try: + if isinstance(data, dict): + for _, obj in data.items(): + if "uuid" in obj: + lookup[obj["uuid"]] = { + "uuid": obj["uuid"], + "display_name": obj["display_name"], + "change": obj["change"], + "attributes": obj["attributes"], + } + if children := obj.get("children"): + _create_diff_lookup(children, lookup) + except Exception: + logger.exception("Cannot create diff lookup") + return lookup + + +def _finalize(markup: t.Any) -> object: + markup = markupsafe.escape(markup) + return capellambse.helpers.replace_hlinks(markup, state.model, _make_href) + + +def _make_href( + obj: m.ModelElement | m.AbstractDiagram, +) -> str | None: + if state.templates_index is None: + return None + + for idx, template in state.templates_index.flat.items(): + if "type" in dir(template.scope): + clsname = template.scope.type + if obj.xtype.rsplit(":", 1)[-1] == clsname: + return f"{ROUTE_PREFIX}/{idx}/{obj.uuid}" + return f"{ROUTE_PREFIX}/__generic__/{obj.uuid}" + + +def _make_href_filter(obj: object) -> str | None: + if jinja2.is_undefined(obj) or obj is None: + return "#" + + if isinstance(obj, m.ElementList): + raise TypeError("Cannot make an href to a list of elements") + if not isinstance(obj, m.ModelElement | m.AbstractDiagram): + raise TypeError(f"Expected a model object, got {obj!r}") + + try: + state.model.by_uuid(obj.uuid) + except KeyError: + return "#" + + return _make_href(obj) + + +def _render_instance_page(template_text, base, object=None): + try: + # render the template with the object + template = state.jinja_env.from_string(template_text) + rendered = template.render( + object=object, + model=state.model, + diff_data=state.diff, + object_diff=state.object_diff, + ) + return fastapi.responses.HTMLResponse( + content=rendered, status_code=200 + ) + except jinja2.TemplateSyntaxError as e: + error_message = markupsafe.Markup( + "Template syntax error: {}, line {}
" + ).format(e.message, e.lineno) + base.error = error_message + print(base) + return fastapi.responses.HTMLResponse(content=error_message) + except Exception as e: + logger.exception("Error rendering template") + trace = markupsafe.escape(traceback.format_exc()) + error_message = markupsafe.Markup( + '' + f"Unexpected error: {type(e).__name__}: {e}" + '
'
+ f"object={object!r}\nmodel={state.model!r}"
+ f"\n\n{trace}"
+ ""
+ )
+ return fastapi.responses.HTMLResponse(content=error_message)
+
+
+app = fastapi.FastAPI(version=capella_model_explorer.__version__)
+app.add_middleware(
+ fastapi.middleware.cors.CORSMiddleware,
+ allow_origins=["*"],
+ allow_credentials=True,
+ allow_methods=["*"],
+ allow_headers=["*"],
+)
+router = fastapi.APIRouter(prefix=ROUTE_PREFIX)
+app.mount(
+ f"{ROUTE_PREFIX}/assets",
+ fastapi.staticfiles.StaticFiles(directory=FRONTEND_DIR / "assets"),
+)
+app.mount(
+ f"{ROUTE_PREFIX}/static",
+ fastapi.staticfiles.StaticFiles(directory=FRONTEND_DIR / "static"),
+)
+
+try:
+ MODEL_INFO = os.environ["MODEL"]
+except KeyError as err:
+ raise SystemExit("MODEL environment variable is not set") from err
+
+try:
+ _ = state.model
+except AttributeError:
+ state.model = capellambse.loadcli(MODEL_INFO)
+
+state.templates = fastapi.templating.Jinja2Templates(directory=FRONTEND_DIR)
+
+state.templates_loader = tl.TemplateLoader(state.model)
+state.jinja_env = jinja2.Environment(
+ loader=jinja2.FileSystemLoader(state.templates_path)
+)
+state.jinja_env.finalize = _finalize
+state.jinja_env.filters["make_href"] = _make_href_filter
+state.templates_index = state.templates_loader.index_path(state.templates_path)
+
+
+@app.middleware("http")
+async def update_last_interaction_time(request: fastapi.Request, call_next):
+ if request.url.path not in ("/metrics", "/favicon.ico"):
+ state.last_interaction = time.time()
+ return await call_next(request)
+
+
+@app.get("/metrics")
+def metrics():
+ idle_time_minutes = (time.time() - state.last_interaction) / 60
+ state.idle_time_gauge.set(idle_time_minutes)
+ return fastapi.Response(
+ content=prometheus_client.generate_latest(),
+ media_type="text/plain",
+ )
+
+
+@router.get("/api/commits")
+async def get_commits():
+ try:
+ return model_diff.populate_commits(state.model)
+ except Exception as e:
+ return {"error": str(e)}
+
+
+@router.post("/api/compare")
+async def post_compare(commit_range: CommitRange):
+ try:
+ state.diff = model_diff.get_diff_data(
+ state.model, commit_range.head, commit_range.prev
+ )
+ state.diff["lookup"] = _create_diff_lookup(state.diff["objects"])
+ if state.diff["lookup"]:
+ return {"success": True}
+ return {"success": False, "error": "No model changes to show"}
+ except Exception as e:
+ logger.exception("Failed to compare versions")
+ return {"success": False, "error": str(e)}
+
+
+@router.get("/api/diff")
+async def get_diff():
+ if state.diff:
+ return state.diff
+ return {"error": "No data available. Please compare two commits first."}
+
+
+@router.get("/api/metadata")
+async def version():
+ return {"version": app.version}
+
+
+@router.get("/api/model-info")
+def model_info():
+ info = state.model.info
+ resinfo = info.resources["\x00"]
+ return {
+ "title": info.title,
+ "revision": resinfo.revision,
+ "hash": resinfo.rev_hash,
+ "capella_version": info.capella_version,
+ "branch": resinfo.branch,
+ "badge": state.model.description_badge,
+ }
+
+
+@router.post("/api/object-diff")
+async def post_object_diff(object_id: ObjectDiffID):
+ if object_id.uuid not in state.diff["lookup"]:
+ raise fastapi.HTTPException(status_code=404, detail="Object not found")
+
+ state.object_diff = state.diff["lookup"][object_id.uuid]
+ return {"success": True}
+
+
+@router.get("/api/objects/{uuid}")
+def read_object(uuid: str):
+ obj = state.model.by_uuid(uuid)
+ details = tl.simple_object(obj)
+ return {
+ "idx": details["idx"],
+ "name": details["name"],
+ "type": obj.xtype,
+ }
+
+
+@router.get("/api/views")
+def read_templates():
+ state.templates_index = state.templates_loader.index_path(
+ state.templates_path
+ )
+ return state.templates_index.as_dict
+
+
+@router.get("/api/views/{template_name}")
+def read_template(template_name: str):
+ template_name = urllib.parse.unquote(template_name)
+ if (
+ state.templates_index is None
+ or template_name not in state.templates_index.flat
+ ):
+ return {
+ "error": (
+ f"Template {template_name} not found"
+ " or templates index not initialized"
+ )
+ }
+ base = state.templates_index.flat[template_name]
+ base.compute_instance_list(state.model)
+ return base
+
+
+@router.get("/api/views/{template_name}/{object_id}")
+def render_template(template_name: str, object_id: str):
+ content = None
+ object = None
+ try:
+ template_name = urllib.parse.unquote(template_name)
+ if (
+ state.templates_index is None
+ or template_name not in state.templates_index.flat
+ ):
+ return {"error": f"Template {template_name} not found"}
+ base = state.templates_index.flat[template_name]
+ template_filename = base.template
+ # load the template file from the templates folder
+ content = (state.templates_path / template_filename).read_text(
+ encoding="utf8"
+ )
+ except Exception as e:
+ error_message = markupsafe.Markup(
+ "Template not found: {}
" + ).format(str(e)) + return fastapi.responses.HTMLResponse(content=error_message) + if object_id == "render": + object = None + else: + try: + object = state.model.by_uuid(object_id) + except Exception as e: + error_message = markupsafe.Markup( + "Requested object not found: {}
" + ).format(str(e)) + return fastapi.responses.HTMLResponse(content=error_message) + return _render_instance_page(content, base, object) + + +# NOTE: Next endpoint (catch-all route) must be located after all other routes! +@router.get("/{rest_of_path:path}") +async def catch_all(request: fastapi.Request, rest_of_path: str): + del rest_of_path + return state.templates.TemplateResponse("index.html", {"request": request}) + + +app.include_router(router) diff --git a/capella_model_explorer/backend/state.py b/capella_model_explorer/backend/state.py new file mode 100644 index 0000000..4bc2cb4 --- /dev/null +++ b/capella_model_explorer/backend/state.py @@ -0,0 +1,34 @@ +# Copyright DB InfraGO AG and contributors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import os +import pathlib +import time +import typing as t + +import prometheus_client + +if t.TYPE_CHECKING: + import capellambse + import fastapi.templating + import jinja2 + + import capella_model_explorer.backend.templates + +diff: dict = {} +idle_time_gauge = prometheus_client.Gauge( + "idletime_minutes", + "Time in minutes since the last user interaction", +) +jinja_env: jinja2.Environment +last_interaction = time.time() +model: capellambse.MelodyModel +object_diff: dict = {} +templates: fastapi.templating.Jinja2Templates +templates_loader: capella_model_explorer.backend.templates.TemplateLoader +templates_index: ( + capella_model_explorer.backend.templates.TemplateCategories | None +) +templates_path = pathlib.Path(os.getenv("TEMPLATES_DIR", "templates")) diff --git a/entrypoint.sh b/entrypoint.sh index 3ea167d..dd5b719 100755 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -7,4 +7,6 @@ sed -i "s|__ROUTE_PREFIX__|${ROUTE_PREFIX}|g" ./frontend/dist/static/env.js sed -i "s|href=\"/|href=\"${ROUTE_PREFIX}/|g" ./frontend/dist/index.html sed -i "s|src=\"/|src=\"${ROUTE_PREFIX}/|g" ./frontend/dist/index.html -exec python -m capella_model_explorer.backend ${MODEL_ENTRYPOINT} /views +export MODEL="${MODEL_ENTRYPOINT}" +export TEMPLATES_DIR="/views" +exec uvicorn --host=0.0.0.0 --port=8000 capella_model_explorer.backend.main:app diff --git a/tests/test_capella_model_explorer.py b/tests/test_capella_model_explorer.py index 0a0030b..11b00ed 100644 --- a/tests/test_capella_model_explorer.py +++ b/tests/test_capella_model_explorer.py @@ -29,7 +29,7 @@ def test_template_loading(): "description": "This is a test template", "scope": {"type": "System", "below": "System"}, } - Template(**template_raw) + Template.model_validate(template_raw) def test_category_loading(): @@ -44,7 +44,7 @@ def test_category_loading(): "idx": "test", "templates": [template_raw], } - TemplateCategory(**category_raw) + TemplateCategory.model_validate(category_raw) def test_index_templates():