Decrease amount of requests

capella2polarion/__main__.py

@@ -83,45 +83,68 @@ def _get_roles_from_config(ctx: dict[str, t.Any]) -> dict[str, list[str]]:
 
 
 def _sanitize_config(
-    config: dict[str, list[str | dict[str, t.Any]]], special: dict[str, t.Any]
+    config: dict[str, list[str | dict[str, t.Any]]],
+    special: list[str | dict[str, t.Any]],
 ) -> dict[str, t.Any]:
+    special_config: dict[str, t.Any] = {}
+    for typ in special:
+        if isinstance(typ, str):
+            special_config[typ] = None
+        else:
+            special_config.update(typ)
+
+    lookup: dict[str, dict[str, list[str]]] = {}
+    for layer, xtypes in config.items():
+        for xt in xtypes:
+            if isinstance(xt, str):
+                item: dict[str, list[str]] = {xt: []}
+            else:
+                item = xt
+
+            lookup.setdefault(layer, {}).update(item)
+
     new_config: dict[str, t.Any] = {}
     for layer, xtypes in config.items():
         new_entries: list[str | dict[str, t.Any]] = []
         for xtype in xtypes:
             if isinstance(xtype, dict):
                 for sub_key, sub_value in xtype.items():
                     new_value = (
-                        special.get("*", [])
-                        + special.get(sub_key, [])
+                        special_config.get("*", [])
+                        + special_config.get(sub_key, [])
                         + sub_value
                     )
                     new_entries.append({sub_key: new_value})
             else:
-                if new_value := special.get("*", []) + special.get(xtype, []):
+                star = special_config.get("*", [])
+                special_xtype = special_config.get(xtype, [])
+                if new_value := star + special_xtype:
                     new_entries.append({xtype: new_value})
                 else:
                     new_entries.append(xtype)
+
+        wildcard_values = special_config.get("*", [])
+        for key, value in special_config.items():
+            if key == "*":
+                continue
+
+            if isinstance(value, list):
+                new_value = (
+                    lookup.get(layer, {}).get(key, [])
+                    + wildcard_values
+                    + value
+                )
+                new_entries.append({key: new_value})
+            elif value is None and key not in [
+                entry if isinstance(entry, str) else list(entry.keys())[0]
+                for entry in new_entries
+            ]:
+                new_entries.append({key: wildcard_values})
         new_config[layer] = new_entries
 
     return new_config
 
 
-def get_polarion_wi_map(
-    ctx: dict[str, t.Any], type_: str = ""
-) -> dict[str, t.Any]:
-    """Return a map from Capella UUIDs to Polarion work items."""
-    types_ = map(elements.helpers.resolve_element_type, ctx.get("TYPES", []))
-    work_item_types = [type_] if type_ else list(types_)
-    _type = " ".join(work_item_types)
-    work_items = ctx["API"].get_all_work_items(
-        f"type:({_type})", {"workitems": "id,uuid_capella,status"}
-    )
-    return {
-        wi.uuid_capella: wi for wi in work_items if wi.id and wi.uuid_capella
-    }
-
-
 @click.group()
 @click.option("--debug/--no-debug", is_flag=True, default=False)
 @click.option("--project-id", required=True, type=str)
@@ -144,12 +167,14 @@ def cli(
         polarion_api_endpoint=f"{ctx.obj['POLARION_HOST']}/rest/v1",
         polarion_access_token=os.environ["POLARION_PAT"],
         custom_work_item=serialize.CapellaWorkItem,
+        add_work_item_checksum=True,
     )
     if not ctx.obj["API"].project_exists():
         sys.exit(1)
 
 
 @cli.command()
+@click.argument("model", type=cli_helpers.ModelCLI())
 @click.argument(
     "diagram_cache",
     type=click.Path(
@@ -160,10 +185,16 @@ def cli(
         path_type=pathlib.Path,
     ),
 )
+@click.argument("config_file", type=click.File(mode="r", encoding="utf8"))
 @click.pass_context
-def diagrams(ctx: click.core.Context, diagram_cache: pathlib.Path) -> None:
-    """Synchronise diagrams."""
-    logger.debug(
+def model_elements(
+    ctx: click.core.Context,
+    model: capellambse.MelodyModel,
+    diagram_cache: pathlib.Path,
+    config_file: t.TextIO,
+) -> None:
+    """Synchronise model elements."""
+    logger.info(
         "Synchronising diagrams from diagram cache at '%s' "
         "to Polarion project with id %r...",
         diagram_cache,
@@ -176,29 +207,12 @@ def diagrams(ctx: click.core.Context, diagram_cache: pathlib.Path) -> None:
 
     ctx.obj["DIAGRAM_CACHE"] = diagram_cache
     ctx.obj["DIAGRAM_IDX"] = json.loads(idx_file.read_text(encoding="utf8"))
-    ctx.obj["CAPELLA_UUIDS"] = [
-        d["uuid"] for d in ctx.obj["DIAGRAM_IDX"] if d["success"]
-    ]
-    ctx.obj["POLARION_WI_MAP"] = get_polarion_wi_map(ctx.obj, "diagram")
-    ctx.obj["POLARION_ID_MAP"] = {
-        uuid: wi.id for uuid, wi in ctx.obj["POLARION_WI_MAP"].items()
-    }
-
-    elements.delete_work_items(ctx.obj)
-    elements.diagram.update_diagrams(ctx.obj)
-    elements.diagram.create_diagrams(ctx.obj)
 
-
-@cli.command()
-@click.argument("model", type=cli_helpers.ModelCLI())
-@click.argument("config_file", type=click.File(mode="r", encoding="utf8"))
-@click.pass_context
-def model_elements(
-    ctx: click.core.Context,
-    model: capellambse.MelodyModel,
-    config_file: t.TextIO,
-) -> None:
-    """Synchronise model elements."""
+    logger.info(
+        "Synchronising model elements (%r) to Polarion project with id %r...",
+        str(elements.ELEMENTS_IDX_PATH),
+        ctx.obj["PROJECT_ID"],
+    )
     ctx.obj["MODEL"] = model
     ctx.obj["CONFIG"] = yaml.safe_load(config_file)
     ctx.obj["ROLES"] = _get_roles_from_config(ctx.obj)
@@ -208,39 +222,26 @@ def model_elements(
     ) = elements.get_elements_and_type_map(ctx.obj)
     ctx.obj["CAPELLA_UUIDS"] = set(ctx.obj["POLARION_TYPE_MAP"])
     ctx.obj["TYPES"] = elements.get_types(ctx.obj)
-    ctx.obj["POLARION_WI_MAP"] = get_polarion_wi_map(ctx.obj)
+    ctx.obj["POLARION_WI_MAP"] = elements.get_polarion_wi_map(ctx.obj)
     ctx.obj["POLARION_ID_MAP"] = {
         uuid: wi.id for uuid, wi in ctx.obj["POLARION_WI_MAP"].items()
     }
+    duuids = {
+        diag["uuid"] for diag in ctx.obj["DIAGRAM_IDX"] if diag["success"]
+    }
+    ctx.obj["ELEMENTS"]["Diagram"] = [
+        diag for diag in ctx.obj["ELEMENTS"]["Diagram"] if diag.uuid in duuids
+    ]
 
-    elements.delete_work_items(ctx.obj)
-    elements.element.update_work_items(ctx.obj)
     elements.element.create_work_items(ctx.obj)
+    elements.delete_work_items(ctx.obj)
+    elements.post_work_items(ctx.obj)
 
-    ctx.obj["POLARION_WI_MAP"] = get_polarion_wi_map(ctx.obj)
-    ctx.obj["POLARION_ID_MAP"] = {
-        uuid: wi.id for uuid, wi in ctx.obj["POLARION_WI_MAP"].items()
-    }
-    elements.element.update_links(ctx.obj)
-
-    diagram_work = get_polarion_wi_map(ctx.obj, "diagram")
-    ctx.obj["POLARION_ID_MAP"] |= {
-        uuid: wi.id for uuid, wi in diagram_work.items()
-    }
-    _diagrams = [
-        diagram
-        for diagram in model.diagrams
-        if diagram.uuid in ctx.obj["POLARION_ID_MAP"]
-    ]
-    ctx.obj["ROLES"]["Diagram"] = ["diagram_elements"]
-    elements.element.update_links(ctx.obj, _diagrams)
+    # Create missing links b/c of unresolved references
+    elements.element.create_work_items(ctx.obj)
+    elements.patch_work_items(ctx.obj)
 
-    elements_index_file = elements.make_model_elements_index(ctx.obj)
-    logger.debug(
-        "Synchronising model objects (%r) to Polarion project with id %r...",
-        str(elements_index_file),
-        ctx.obj["PROJECT_ID"],
-    )
+    elements.make_model_elements_index(ctx.obj)
 
 
 if __name__ == "__main__":

capella2polarion/elements/__init__.py

@@ -12,6 +12,7 @@
     "STATUS_DELETE",
 ]
 
+import functools
 import logging
 import pathlib
 import typing as t
@@ -20,6 +21,7 @@
 import polarion_rest_api_client as polarion_api
 import yaml
 from capellambse.model import common
+from capellambse.model import diagram as diag
 
 logger = logging.getLogger(__name__)
 
@@ -34,7 +36,7 @@
     "PhysicalComponentNode": "PhysicalComponent",
     "PhysicalComponentBehavior": "PhysicalComponent",
 }
-POL2CAPELLA_TYPES = (
+POL2CAPELLA_TYPES: dict[str, str] = (
     {
         "OperationalEntity": "Entity",
         "OperationalInteraction": "FunctionalExchange",
@@ -45,6 +47,21 @@
 )
 
 
+def get_polarion_wi_map(
+    ctx: dict[str, t.Any], type_: str = ""
+) -> dict[str, t.Any]:
+    """Return a map from Capella UUIDs to Polarion work items."""
+    types_ = map(helpers.resolve_element_type, ctx.get("TYPES", []))
+    work_item_types = [type_] if type_ else list(types_)
+    _type = " ".join(work_item_types)
+    work_items = ctx["API"].get_all_work_items(
+        f"type:({_type})", {"workitems": "id,uuid_capella,checksum,status"}
+    )
+    return {
+        wi.uuid_capella: wi for wi in work_items if wi.id and wi.uuid_capella
+    }
+
+
 def delete_work_items(ctx: dict[str, t.Any]) -> None:
     """Delete work items in a Polarion project.
 
@@ -58,7 +75,7 @@ def delete_work_items(ctx: dict[str, t.Any]) -> None:
     """
 
     def serialize_for_delete(uuid: str) -> str:
-        logger.debug(
+        logger.info(
             "Delete work item %r...",
             workitem_id := ctx["POLARION_ID_MAP"][uuid],
         )
@@ -78,6 +95,78 @@ def serialize_for_delete(uuid: str) -> str:
             logger.error("Deleting work items failed. %s", error.args[0])
 
 
+def post_work_items(ctx: dict[str, t.Any]) -> None:
+    """Post work items in a Polarion project.
+
+    Parameters
+    ----------
+    ctx
+        The context for the workitem operation to be processed.
+    """
+    work_items: list[serialize.CapellaWorkItem] = []
+    for work_item in ctx["WORK_ITEMS"].values():
+        if work_item.uuid_capella in ctx["POLARION_ID_MAP"]:
+            continue
+
+        assert work_item is not None
+        work_items.append(work_item)
+        logger.info("Create work item for %r...", work_item.title)
+    if work_items:
+        try:
+            ctx["API"].create_work_items(work_items)
+            workitems = {wi.uuid_capella: wi for wi in work_items if wi.id}
+            ctx["POLARION_WI_MAP"].update(workitems)
+            ctx["POLARION_ID_MAP"] = {
+                uuid: wi.id for uuid, wi in ctx["POLARION_WI_MAP"].items()
+            }
+        except polarion_api.PolarionApiException as error:
+            logger.error("Creating work items failed. %s", error.args[0])
+
+
+def patch_work_items(ctx: dict[str, t.Any]) -> None:
+    """Update work items in a Polarion project.
+
+    Parameters
+    ----------
+    ctx
+        The context for the workitem operation to be processed.
+    """
+
+    def add_content(
+        obj: common.GenericElement | diag.Diagram,
+        ctx: dict[str, t.Any],
+        **kwargs,
+    ) -> serialize.CapellaWorkItem:
+        work_item = ctx["WORK_ITEMS"][obj.uuid]
+        for key, value in kwargs.items():
+            if getattr(work_item, key, None) is None:
+                continue
+
+            setattr(work_item, key, value)
+        return work_item
+
+    ctx["POLARION_ID_MAP"] = uuids = {
+        uuid: wi.id
+        for uuid, wi in ctx["POLARION_WI_MAP"].items()
+        if wi.status == "open" and wi.uuid_capella and wi.id
+    }
+    for uuid in uuids:
+        elements = ctx["MODEL"]
+        if uuid.startswith("_"):
+            elements = ctx["MODEL"].diagrams
+        obj = elements.by_uuid(uuid)
+
+        links = element.create_links(obj, ctx)
+
+        api_helper.patch_work_item(
+            ctx,
+            obj,
+            functools.partial(add_content, linked_work_items=links),
+            obj._short_repr_(),
+            "element",
+        )
+
+
 def get_types(ctx: dict[str, t.Any]) -> set[str]:
     """Return a set of Polarion types from the current context."""
     xtypes = set[str]()
@@ -100,13 +189,24 @@ def get_elements_and_type_map(
             if isinstance(typ, dict):
                 typ = list(typ.keys())[0]
 
+            if typ == "Diagram":
+                continue
+
             xtype = convert_type.get(typ, typ)
             objects = ctx["MODEL"].search(xtype, below=below)
             elements.setdefault(typ, []).extend(objects)
             for obj in objects:
                 type_map[obj.uuid] = typ
 
     _fix_components(elements, type_map)
+    diagrams_from_cache = {
+        d["uuid"] for d in ctx["DIAGRAM_IDX"] if d["success"]
+    }
+    elements["Diagram"] = [
+        d for d in ctx["MODEL"].diagrams if d.uuid in diagrams_from_cache
+    ]
+    for obj in elements["Diagram"]:
+        type_map[obj.uuid] = "Diagram"
     return elements, type_map
 
 
@@ -150,7 +250,7 @@ def _fix_components(
         elements["PhysicalComponent"] = components
 
 
-def make_model_elements_index(ctx: dict[str, t.Any]) -> pathlib.Path:
+def make_model_elements_index(ctx: dict[str, t.Any]) -> None:
     """Create an elements index file for all migrated elements."""
     elements: list[dict[str, t.Any]] = []
     for obj in chain.from_iterable(ctx["ELEMENTS"].values()):
@@ -176,7 +276,11 @@ def make_model_elements_index(ctx: dict[str, t.Any]) -> pathlib.Path:
         elements.append(element_)
 
     ELEMENTS_IDX_PATH.write_text(yaml.dump(elements), encoding="utf8")
-    return ELEMENTS_IDX_PATH
 
 
-from . import diagram, element, helpers, serialize
+from . import (  # pylint: disable=cyclic-import
+    api_helper,
+    element,
+    helpers,
+    serialize,
+)