Skip to content

Commit

Permalink
feat(decl): Implement metadata for decl YAML
Browse files Browse the repository at this point in the history
These changes make `capellambse.decl` understand a new metadata section
in the declarative modelling file.

The `decl.apply` function gained a new `strict` argument. If set to
True, metadata must be present in the provided YAML and match the passed
model.
  • Loading branch information
ewuerger authored and Wuestengecko committed Oct 4, 2024
1 parent 8394e9b commit dc55401
Show file tree
Hide file tree
Showing 5 changed files with 473 additions and 15 deletions.
1 change: 1 addition & 0 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ repos:
additional_dependencies:
- mypypp==0.1.1

- awesomeversion==24.2.0
- click==8.1.7
- diskcache==5.0
- jinja2==3.1.3
Expand Down
217 changes: 205 additions & 12 deletions capellambse/decl.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,17 +21,24 @@
"apply",
"dump",
"load",
"load_with_metadata",
]

import collections
import collections.abc as cabc
import contextlib
import dataclasses
import importlib.metadata as imm
import logging
import operator
import os
import pathlib
import re
import sys
import typing as t

import awesomeversion as av
import typing_extensions as te
import yaml

import capellambse
Expand All @@ -45,11 +52,66 @@
"Promise",
capellambse.ModelObject | _FutureAction,
]


def dump(instructions: cabc.Sequence[cabc.Mapping[str, t.Any]]) -> str:
"""Dump an instruction stream to YAML."""
return yaml.dump(instructions, Dumper=YDMDumper)
logger = logging.getLogger(__name__)


class _WriterMetadata(te.TypedDict):
capellambse: str
generator: te.NotRequired[str]


class _ModelMetadata(te.TypedDict):
url: str
revision: te.NotRequired[str]
entrypoint: str


class Metadata(te.TypedDict, total=False):
written_by: _WriterMetadata
model: _ModelMetadata


@t.overload
def dump(
instructions: cabc.Sequence[cabc.Mapping[str, t.Any]],
*,
metadata: Metadata | None = None,
) -> str: ...
@t.overload
def dump(
instructions: cabc.Sequence[cabc.Mapping[str, t.Any]],
*,
metadata: m.MelodyModel,
generator: str | None = None,
) -> str: ...
def dump(
instructions: cabc.Sequence[cabc.Mapping[str, t.Any]],
*,
metadata: m.MelodyModel | Metadata | None = None,
generator: str | None = None,
) -> str:
"""Dump an instruction stream to YAML.
Optionally dump metadata with the instruction stream to YAML.
"""
if isinstance(metadata, m.MelodyModel):
res_info = metadata.info.resources["\x00"]
metadata = {
"model": {
"url": res_info.url,
"revision": res_info.rev_hash,
"entrypoint": str(metadata.info.entrypoint),
},
"written_by": {
"capellambse": capellambse.__version__.split("+", 1)[0]
},
}
if generator is not None:
metadata["written_by"]["generator"] = generator

if not metadata:
return yaml.dump(instructions, Dumper=YDMDumper)
return yaml.dump_all([metadata, instructions], Dumper=YDMDumper)


def load(file: FileOrPath) -> list[dict[str, t.Any]]:
Expand All @@ -58,8 +120,36 @@ def load(file: FileOrPath) -> list[dict[str, t.Any]]:
Parameters
----------
file
An open file-like object, or a path or PathLike pointing to such
a file. Files are expected to use UTF-8 encoding.
An open file-like object containing decl instructions, or a path
or PathLike pointing to such a file. Files are expected to use
UTF-8 encoding.
"""
_, instructions = load_with_metadata(file)
return instructions


def load_with_metadata(
file: FileOrPath,
) -> tuple[Metadata, list[dict[str, t.Any]]]:
"""Load an instruction stream and its metadata from a YAML file.
If the file does not have a metadata section, an empty dict will be
returned.
Parameters
----------
file
An open file-like object containing decl instructions, or a path
or PathLike pointing to such a file. Files are expected to use
UTF-8 encoding.
Returns
-------
dict[str, Any]
The metadata read from the file, or an empty dictionary if the
file did not contain any metadata.
list[dict[str, Any]]
The instruction stream.
"""
if hasattr(file, "read"):
file = t.cast(t.IO[str], file)
Expand All @@ -69,11 +159,24 @@ def load(file: FileOrPath) -> list[dict[str, t.Any]]:
ctx = open(file, encoding="utf-8") # noqa: SIM115

with ctx as opened_file:
return yaml.load(opened_file, Loader=YDMLoader)
contents = list(yaml.load_all(opened_file, Loader=YDMLoader))

if len(contents) == 2:
return (t.cast(Metadata, contents[0]) or {}, contents[1] or [])
if len(contents) == 1:
return ({}, contents[0] or [])
if len(contents) == 0:
return ({}, [])
raise ValueError(
f"Expected a YAML file with 1 or 2 documents, found {len(contents)}"
)


def apply(
model: capellambse.MelodyModel, file: FileOrPath
model: capellambse.MelodyModel,
file: FileOrPath,
*,
strict: bool = False,
) -> dict[Promise, capellambse.ModelObject]:
"""Apply a declarative modelling file to the given model.
Expand All @@ -88,6 +191,9 @@ def apply(
The full format of these files is documented in the
:ref:`section about declarative modelling
<declarative-modelling>`.
strict
Verify metadata contained in the file against the used model,
and raise an error if they don't match.
Notes
-----
Expand All @@ -103,10 +209,18 @@ def apply(
``!promise``, but reorderings are still possible even if no promises
are used in an input document.
"""
instructions = collections.deque(load(file))
metadata, raw_instructions = load_with_metadata(file)
instructions = collections.deque(raw_instructions)
promises = dict[Promise, capellambse.ModelObject]()
deferred = collections.defaultdict[Promise, list[_FutureAction]](list)

try:
_verify_metadata(model, metadata)
except ValueError as err:
if strict:
raise
logger.warning("Metadata does not match provided model: %s", err)

while instructions:
instruction = instructions.popleft()

Expand Down Expand Up @@ -148,6 +262,80 @@ def apply(
return promises


def _verify_metadata(
model: capellambse.MelodyModel, metadata: Metadata
) -> None:
if not metadata:
raise ValueError("Cannot verify decl metadata: No metadata found")

written_by = metadata.get("written_by", {}).get("capellambse", "")
if not written_by:
raise ValueError(
"Unsupported YAML: Can't find 'written_by:capellambse' in metadata"
)
if not _is_pep440(written_by):
raise ValueError(f"Malformed version number in metadata: {written_by}")

current = av.AwesomeVersion(
imm.version("capellambse").partition("+")[0],
ensure_strategy=av.AwesomeVersionStrategy.PEP440,
)
try:
written_version = av.AwesomeVersion(
written_by,
ensure_strategy=av.AwesomeVersionStrategy.PEP440,
)
version_matches = current >= written_version
except Exception as err:
raise ValueError(
"Cannot apply decl: Cannot verify required capellambse version:"
f" {type(err).__name__}: {err}"
) from None

if not version_matches:
raise ValueError(
"Cannot apply decl: This capellambse is too old for this YAML:"
f" Need at least v{written_by}, but have only v{current})"
)

model_metadata = metadata.get("model", {})
res_info = model.info.resources["\x00"]
url = model_metadata.get("url")
if url != res_info.url:
raise ValueError(
"Cannot apply decl: Model URL mismatch:"
f" YAML expects {url}, current is {res_info.url}"
)

hash = model_metadata.get("revision")
if hash != res_info.rev_hash:
raise ValueError(
"Cannot apply decl: Model version mismatch:"
f" YAML expects {hash}, current is {res_info.rev_hash}"
)

entrypoint = pathlib.PurePosixPath(model_metadata.get("entrypoint", ""))
if entrypoint != model.info.entrypoint:
raise ValueError(
"Cannot apply decl: Model entrypoint mismatch:"
f" YAML expects {entrypoint}, current is {model.info.entrypoint}"
)


def _is_pep440(version: str) -> bool:
"""Check if given version aligns with PEP440.
See Also
--------
https://peps.python.org/pep-0440/#appendix-b-parsing-version-strings-with-regular-expressions
"""
pep440_ptrn = re.compile(
r"([1-9][0-9]*!)?(0|[1-9][0-9]*)(\.(0|[1-9][0-9]*))*((a|b|rc)"
r"(0|[1-9][0-9]*))?(\.post(0|[1-9][0-9]*))?(\.dev(0|[1-9][0-9]*))?"
)
return pep440_ptrn.fullmatch(version) is not None


def _operate_create(
promises: dict[Promise, capellambse.ModelObject],
parent: capellambse.ModelObject,
Expand Down Expand Up @@ -614,10 +802,15 @@ def _main() -> None:

@click.command()
@click.option("-m", "--model", type=capellambse.ModelCLI(), required=True)
@click.option("-s", "--strict/--relaxed", is_flag=True, default=False)
@click.argument("file", type=click.File("r"))
def _main(model: capellambse.MelodyModel, file: t.IO[str]) -> None:
def _main(
model: capellambse.MelodyModel,
file: t.IO[str],
strict: bool,
) -> None:
"""Apply a declarative modelling YAML file to a model."""
apply(model, file)
apply(model, file, strict=strict)
model.save()


Expand Down
39 changes: 36 additions & 3 deletions docs/source/start/declarative.rst
Original file line number Diff line number Diff line change
Expand Up @@ -68,9 +68,42 @@ containing YAML, wrap it in :external:class:`io.StringIO`:
Format description
==================

The expected YAML follows a simple format, where a parent object (i.e. an
object that already exists in the model) is selected, and one or more of three
different operations is applied to it:
The YAML file may contain one or two YAML documents (separated by a line
containing only three minus signs ``---``). The first document contains
metadata, while the second document contains the instructions to perform
against the model. The metadata document may be omitted, in which case the file
only contains an instruction stream.

Metadata
--------

.. versionadded:: 0.6.8
Added metadata section to the declarative modelling YAML.

The metadata section is optional and has the following format:

.. code-block:: yaml
model:
url: https://example.com/model.git
revision: 0123456789abcdefdeadbeef0123456789abcdef
entrypoint: path/to/model.aird
written_by:
capellambse_version: 1.0.0
generator: Example Generator 1.0.0
It contains information about which model the declarative modelling YAML file
wants to change, and which capellambse version and generator it was written
with. A versioned model can be uniquely identified by its repository URL, the
revision, and the model entrypoint. ``decl.apply()`` with ``strict=True`` will
verify these values against the ``model.info`` of the passed model.

Instructions
------------

The expected instruction document in the YAML follows a simple format, where a
parent object (i.e. an object that already exists in the model) is selected,
and one or more of three different operations is applied to it:

- ``extend``-ing the object on list attributes,
- ``set``-ting properties on the object itself,
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ classifiers = [
"Typing :: Typed",
]
dependencies = [
"awesomeversion>=24.2.0",
"diskcache>=5.0",
"lxml>=4.5.0",
"markupsafe>=2.0",
Expand Down
Loading

0 comments on commit dc55401

Please sign in to comment.