feat: Add id_prefix to work items

Pull request #62 from DSD-DBS/add-id-prefix

capella2polarion/__main__.py

@@ -40,6 +40,8 @@
     type=click.File(mode="r", encoding="utf8"),
     default=None,
 )
+@click.option("--type-prefix", type=str, default="")
+@click.option("--role-prefix", type=str, default="")
 @click.pass_context
 def cli(
     ctx: click.core.Context,
@@ -51,6 +53,8 @@ def cli(
     polarion_delete_work_items: bool,
     capella_model: capellambse.MelodyModel,
     synchronize_config: typing.TextIO,
+    type_prefix: str,
+    role_prefix: str,
 ) -> None:
     """Synchronise data from Capella to Polarion."""
     if capella_model.diagram_cache is None:
@@ -65,6 +69,8 @@ def cli(
         capella_model,
         synchronize_config,
         force_update,
+        type_prefix,
+        role_prefix,
     )
     capella2polarion_cli.setup_logger()
     ctx.obj = capella2polarion_cli
@@ -92,6 +98,8 @@ def synchronize(ctx: click.core.Context) -> None:
     converter = model_converter.ModelConverter(
         capella_to_polarion_cli.capella_model,
         capella_to_polarion_cli.polarion_params.project_id,
+        type_prefix=capella_to_polarion_cli.type_prefix,
+        role_prefix=capella_to_polarion_cli.role_prefix,
     )
 
     converter.read_model(capella_to_polarion_cli.config)
@@ -100,6 +108,8 @@ def synchronize(ctx: click.core.Context) -> None:
         capella_to_polarion_cli.polarion_params,
         capella_to_polarion_cli.config,
         capella_to_polarion_cli.force_update,
+        type_prefix=capella_to_polarion_cli.type_prefix,
+        role_prefix=capella_to_polarion_cli.role_prefix,
     )
 
     polarion_worker.load_polarion_work_item_map()

capella2polarion/cli.py

@@ -29,6 +29,8 @@ def __init__(
         capella_model: capellambse.MelodyModel,
         synchronize_config_io: typing.TextIO,
         force_update: bool = False,
+        type_prefix: str = "",
+        role_prefix: str = "",
     ) -> None:
         self.debug = debug
         self.polarion_params = pw.PolarionWorkerParams(
@@ -42,6 +44,8 @@ def __init__(
         self.synchronize_config_io: typing.TextIO = synchronize_config_io
         self.config = converter_config.ConverterConfig()
         self.force_update = force_update
+        self.type_prefix = type_prefix
+        self.role_prefix = role_prefix
 
     def _none_save_value_string(self, value: str | None) -> str | None:
         return "None" if value is None else value
@@ -90,19 +94,12 @@ def _value(value):
     def setup_logger(self) -> None:
         """Set the logger in the right mood."""
         max_logging_level = logging.DEBUG if self.debug else logging.WARNING
-        assert isinstance(logger.parent, logging.RootLogger)
-        logger.parent.setLevel(max_logging_level)
-        log_formatter = logging.Formatter(
-            "%(asctime)-15s - %(levelname)-8s %(message)s"
+        logging.basicConfig(
+            level=max_logging_level,
+            format="%(asctime)-15s - %(levelname)-8s %(message)s",
         )
-        console_handler = logging.StreamHandler()
-        console_handler.setLevel(max_logging_level)
-        console_handler.setFormatter(log_formatter)
-        console_handler.addFilter(
-            lambda record: record.name.startswith("capella2polarion")
-            or (record.name == "httpx" and record.levelname == "INFO")
-        )
-        logger.parent.addHandler(console_handler)
+        logging.getLogger("httpx").setLevel("WARNING")
+        logging.getLogger("httpcore").setLevel("WARNING")
 
     def load_synchronize_config(self) -> None:
         """Read the sync config into SynchronizeConfigContent.

capella2polarion/connectors/polarion_worker.py

@@ -47,11 +47,15 @@ def __init__(
         params: PolarionWorkerParams,
         config: converter_config.ConverterConfig,
         force_update: bool = False,
+        type_prefix: str = "",
+        role_prefix: str = "",
     ) -> None:
         self.polarion_params = params
         self.polarion_data_repo = polarion_repo.PolarionDataRepository()
         self.config = config
         self.force_update = force_update
+        self.type_prefix = type_prefix
+        self.role_prefix = role_prefix
 
         if (self.polarion_params.project_id is None) or (
             len(self.polarion_params.project_id) == 0
@@ -92,13 +96,10 @@ def check_client(self) -> None:
 
     def load_polarion_work_item_map(self):
         """Return a map from Capella UUIDs to Polarion work items."""
-        _type = " ".join(self.config.polarion_types)
-
         work_items = self.client.get_all_work_items(
-            f"type:({_type})",
+            "HAS_VALUE:uuid_capella",
             {"workitems": "id,uuid_capella,checksum,status"},
         )
-
         self.polarion_data_repo.update_work_items(work_items)
 
     def delete_work_items(

capella2polarion/converters/element_converter.py

@@ -111,11 +111,13 @@ def __init__(
         capella_polarion_mapping: polarion_repo.PolarionDataRepository,
         converter_session: data_session.ConverterSession,
         generate_attachments: bool,
+        type_prefix: str = "",
     ):
         self.model = model
         self.capella_polarion_mapping = capella_polarion_mapping
         self.converter_session = converter_session
         self.generate_attachments = generate_attachments
+        self.type_prefix = type_prefix
 
     def serialize_all(self) -> list[data_models.CapellaWorkItem]:
         """Serialize all items of the converter_session."""
@@ -130,7 +132,9 @@ def serialize(self, uuid: str) -> data_models.CapellaWorkItem | None:
             uuid
         ):
             work_item_id = old.id
+
         self.__generic_work_item(converter_data, work_item_id)
+        assert converter_data.work_item is not None
 
         assert isinstance(converter_data.type_config.converters, dict)
         for converter, params in converter_data.type_config.converters.items():
@@ -144,6 +148,11 @@ def serialize(self, uuid: str) -> data_models.CapellaWorkItem | None:
                 converter_data.errors.add(error.args[0])
                 converter_data.work_item = None
 
+        if self.type_prefix and converter_data.work_item is not None:
+            converter_data.work_item.type = (
+                f"{self.type_prefix}_{converter_data.work_item.type}"
+            )
+
         if converter_data.errors:
             log_args = (
                 converter_data.capella_element._short_repr_(),
@@ -235,6 +244,7 @@ def _draw_additional_attributes_diagram(
         )
         if attachment:
             self._add_attachment(work_item, attachment)
+
         work_item.additional_attributes[attribute] = {
             "type": "text/html",
             "value": diagram_html,
@@ -368,7 +378,9 @@ def _get_requirement_types_text(
     # Serializer implementation starts below
 
     def __generic_work_item(
-        self, converter_data: data_session.ConverterData, work_item_id
+        self,
+        converter_data: data_session.ConverterData,
+        work_item_id: str | None,
     ) -> data_models.CapellaWorkItem:
         obj = converter_data.capella_element
         raw_description = getattr(obj, "description", None)
@@ -377,14 +389,15 @@ def __generic_work_item(
         )
         converter_data.description_references = uuids
         requirement_types = self._get_requirement_types_text(obj)
+
         converter_data.work_item = data_models.CapellaWorkItem(
             id=work_item_id,
             type=converter_data.type_config.p_type,
             title=obj.name,
+            uuid_capella=obj.uuid,
             description_type="text/html",
             description=value,
             status="open",
-            uuid_capella=obj.uuid,
             **requirement_types,
         )
         for attachment in attachments:
@@ -410,13 +423,14 @@ def _diagram(
             id=work_item_id,
             type=converter_data.type_config.p_type,
             title=diagram.name,
+            uuid_capella=diagram.uuid,
             description_type="text/html",
             description=diagram_html,
             status="open",
-            uuid_capella=diagram.uuid,
         )
         if attachment:
             self._add_attachment(converter_data.work_item, attachment)
+
         return converter_data.work_item
 
     def _include_pre_and_post_condition(
@@ -447,7 +461,6 @@ def _linked_text_as_description(
         self, converter_data: data_session.ConverterData
     ) -> data_models.CapellaWorkItem:
         """Return attributes for a ``Constraint``."""
-        # pylint: disable-next=attribute-defined-outside-init
         assert converter_data.work_item, "No work item set yet"
         (
             uuids,

capella2polarion/converters/link_converter.py

@@ -40,11 +40,13 @@ def __init__(
         converter_session: data_session.ConverterSession,
         project_id: str,
         model: capellambse.MelodyModel,
+        role_prefix: str = "",
     ):
         self.capella_polarion_mapping = capella_polarion_mapping
         self.converter_session = converter_session
         self.project_id = project_id
         self.model = model
+        self.role_prefix = role_prefix
 
         self.serializers: dict[str, _Serializer] = {
             "description_reference": self._handle_description_reference_links,
@@ -59,42 +61,59 @@ def create_links_for_work_item(
         """Create work item links for a given Capella object."""
         converter_data = self.converter_session[uuid]
         obj = converter_data.capella_element
-        if isinstance(obj, diag.Diagram):
-            repres = f"<Diagram {obj.name!r}>"
-        else:
-            repres = obj._short_repr_()
-
         work_item = converter_data.work_item
         assert work_item is not None
         new_links: list[polarion_api.WorkItemLink] = []
+        link_errors: list[str] = []
         for link_config in converter_data.type_config.links:
             assert (role_id := link_config.polarion_role) is not None
             attr_id = link_config.capella_attr or ""
-            if serializer := self.serializers.get(role_id):
-                new_links.extend(
-                    serializer(obj, work_item.id, role_id, attr_id, {})
-                )
-            else:
-                if (refs := _resolve_attribute(obj, attr_id)) is None:
-                    logger.info(
-                        "Unable to create work item link %r for [%s]. "
-                        "There is no %r attribute on %s or no link-serializer",
-                        role_id,
-                        work_item.id,
-                        attr_id,
-                        repres,
+            serializer = self.serializers.get(role_id)
+            if self.role_prefix:
+                role_id = f"{self.role_prefix}_{role_id}"
+            try:
+                if serializer:
+                    new_links.extend(
+                        serializer(obj, work_item.id, role_id, attr_id, {})
                     )
-                    continue
-
-                new: cabc.Iterable[str]
-                if isinstance(refs, common.ElementList):
-                    new = refs.by_uuid  # type: ignore[assignment]
                 else:
-                    assert hasattr(refs, "uuid")
-                    new = [refs.uuid]
+                    refs = _resolve_attribute(obj, attr_id)
+                    new: cabc.Iterable[str]
+                    if isinstance(refs, common.ElementList):
+                        new = refs.by_uuid  # type: ignore[assignment]
+                    else:
+                        assert hasattr(refs, "uuid"), "No 'uuid' on value"
+                        new = [refs.uuid]
+
+                    new = set(
+                        self._get_work_item_ids(work_item.id, new, role_id)
+                    )
+                    new_links.extend(
+                        self._create(work_item.id, role_id, new, {})
+                    )
+            except Exception as error:
+                request_text = f"Requested attribute: {attr_id}"
+                error_text = f"{type(error).__name__} {str(error)}"
+                link_errors.extend([request_text, error_text, "--------"])
+
+        if link_errors:
+            for link_error in link_errors:
+                converter_data.errors.add(link_error)
+
+            log_args = (
+                converter_data.capella_element._short_repr_(),
+                "\n\t".join(link_errors),
+            )
+            if not new_links:
+                logger.error("Link creation for %r failed:\n\t%s", *log_args)
+            else:
+                logger.warning(
+                    "Link creation for %r partially successful. Some links "
+                    "were not created:"
+                    "\n\t%s",
+                    *log_args,
+                )
 
-                new = set(self._get_work_item_ids(work_item.id, new, role_id))
-                new_links.extend(self._create(work_item.id, role_id, new, {}))
         return new_links
 
     def _get_work_item_ids(
@@ -193,7 +212,6 @@ def _handle_exchanges(
     ) -> list[polarion_api.WorkItemLink]:
         exchanges: list[str] = []
         objs = _resolve_attribute(obj, attr_id)
-        assert objs is not None
         exs = self._get_work_item_ids(work_item_id, objs.by_uuid, role_id)
         exchanges.extend(set(exs))
         return self._create(work_item_id, role_id, exchanges, links)
@@ -233,7 +251,10 @@ def create_grouped_link_fields(
                     break
 
             self._create_link_fields(
-                work_item, role, grouped_links, config=config
+                work_item,
+                role.removeprefix(f"{self.role_prefix}_"),
+                grouped_links,
+                config=config,
             )
 
     def _create_link_fields(
@@ -316,7 +337,12 @@ def create_grouped_back_link_fields(
             List of links referencing work_item as secondary
         """
         for role, grouped_links in _group_by("role", links).items():
-            self._create_link_fields(work_item, role, grouped_links, True)
+            self._create_link_fields(
+                work_item,
+                role.removeprefix(f"{self.role_prefix}_"),
+                grouped_links,
+                True,
+            )
 
 
 def _group_by(
@@ -363,9 +389,9 @@ def _sorted_unordered_html_list(
 
 def _resolve_attribute(
     obj: common.GenericElement, attr_id: str
-) -> common.ElementList[common.GenericElement] | None:
+) -> common.ElementList[common.GenericElement]:
     attr_name, _, map_id = attr_id.partition(".")
-    objs = getattr(obj, attr_name, None)
-    if map_id and objs is not None:
+    objs = getattr(obj, attr_name)
+    if map_id:
         objs = objs.map(map_id)
     return objs