implemented POM adapter for somesy

Author: apirogov

src/somesy/codemeta/writer.py

@@ -167,4 +167,6 @@ def sync(self, metadata: ProjectMetadata) -> None:
         Use existing sync function from ProjectMetadataWriter but update repository and contributors.
         """
         super().sync(metadata)
-        self.contributors = self._sync_person_list(self.contributors, metadata.people)
+        self.contributors = self._sync_person_list(
+            self.contributors, metadata.contributors()
+        )

src/somesy/core/models.py

@@ -36,6 +36,7 @@ class SomesyBaseModel(BaseModel):
 
     model_config = dict(
         extra="forbid",
+        validate_assignment=True,
         populate_by_name=True,
         str_strip_whitespace=True,
         str_min_length=1,

src/somesy/core/writer.py

@@ -49,7 +49,7 @@ def __init__(
             direct_mappings: Dict with direct mappings of keys between somesy and target
         """
         self._data: DictLike = {}
-        self.path = path
+        self.path = path if isinstance(path, Path) else Path(path)
         self.create_if_not_exists = create_if_not_exists
         self.direct_mappings = direct_mappings or {}
 
@@ -100,7 +100,11 @@ def save(self, path: Optional[Path]) -> None:
         """
 
     def _get_property(
-        self, key: Union[str, List[str]], *, remove: bool = False
+        self,
+        key: Union[str, List[str]],
+        *,
+        only_first: bool = False,
+        remove: bool = False,
     ) -> Optional[Any]:
         """Get a property from the data.
 
@@ -109,6 +113,7 @@ def _get_property(
 
         Args:
             key: Name of the key or sequence of multiple keys to retrieve the value.
+            only_first: If True, returns only first entry if the value is a list.
             remove: If True, will remove the retrieved value and clean up the dict.
         """
         key_path = [key] if isinstance(key, str) else key
@@ -117,6 +122,7 @@ def _get_property(
         seq = [curr]
         for k in key_path:
             curr = curr.get(k)
+            curr = curr[0] if isinstance(curr, list) and only_first else curr
             seq.append(curr)
             if curr is None:
                 return None
@@ -129,11 +135,15 @@ def _get_property(
                 if not dct.get(key):
                     del dct[key]
 
+        if isinstance(curr, list) and only_first:
+            return curr[0]
         return curr
 
     def _set_property(self, key: Union[str, List[str], IgnoreKey], value: Any) -> None:
         """Set a property in the data.
 
+        Note if there are lists along the path, they are cleared out.
+
         Override this to e.g. rewrite the retrieved key
         (e.g. if everything relevant is in some subobject).
         """
@@ -151,6 +161,7 @@ def _set_property(self, key: Union[str, List[str], IgnoreKey], value: Any) -> No
             if key not in curr:
                 curr[key] = {}
             curr = curr[key]
+
         curr[key_path[-1]] = value
 
     # ----
@@ -253,6 +264,9 @@ def sync(self, metadata: ProjectMetadata) -> None:
         self.maintainers = self._sync_person_list(
             self.maintainers, metadata.maintainers()
         )
+        # self.contributors = self._sync_person_list(
+        #     self.contributors, metadata.contributors()
+        # )
 
         self.license = metadata.license.value
 
@@ -275,8 +289,6 @@ def _to_person(person_obj: Any) -> Person:
     @classmethod
     def _parse_people(cls, people: Optional[List[Any]]) -> List[Person]:
         """Return a list of Persons parsed from list of format-specific people representations."""
-        if not people:
-            return []
         return list(map(cls._to_person, people or []))
 
     # ----

src/somesy/package_json/writer.py

@@ -2,7 +2,7 @@
 import logging
 from collections import OrderedDict
 from pathlib import Path
-from typing import List, Optional
+from typing import Dict, List, Optional, Union
 
 from rich.pretty import pretty_repr
 
@@ -112,5 +112,19 @@ def sync(self, metadata: ProjectMetadata) -> None:
         """
         super().sync(metadata)
         self.contributors = self._sync_person_list(self.contributors, metadata.people)
-        if metadata.repository:
-            self.repository = {"type": "git", "url": str(metadata.repository)}
+
+    @property
+    def repository(self) -> Optional[Union[str, Dict]]:
+        """Return the repository url of the project."""
+        if repo := super().repository:
+            if isinstance(repo, str):
+                return repo
+            else:
+                return repo.get("url")
+        else:
+            return None
+
+    @repository.setter
+    def repository(self, value: Optional[Union[str, Dict]]) -> None:
+        """Set the repository url of the project."""
+        self._set_property(self._get_key("repository"), dict(type="git", url=value))

src/somesy/pom_xml/__init__.py

@@ -3,7 +3,6 @@
 # some POM-related constants and reusable objects
 POM_URL = "http://maven.apache.org/POM/4.0.0"
 POM_PREF = "{" + POM_URL + "}"
-POM_NS_MAP = dict(pom=POM_URL)
 POM_ROOT_ATRS = {
     "xmlns": "http://maven.apache.org/POM/4.0.0",
     "xmlns:xsi": "http://www.w3.org/2001/XMLSchema-instance",

src/somesy/pom_xml/writer.py

@@ -2,15 +2,14 @@
 import logging
 import xml.etree.ElementTree as ET
 from pathlib import Path
-from typing import Optional
+from typing import Any, Dict, List, Optional, Union
 
 from somesy.core.models import Person
 from somesy.core.writer import FieldKeyMapping, ProjectMetadataWriter
 
 from . import POM_ROOT_ATRS, POM_URL
 from .xmlproxy import XMLProxy
 
-ET.register_namespace("pom", POM_URL)  # globally register xml namespace for POM
 logger = logging.getLogger("somesy")
 
 
@@ -30,10 +29,12 @@ def __init__(
         See [somesy.core.writer.ProjectMetadataWriter.__init__][].
         """
         mappings: FieldKeyMapping = {
-            "year": ["inceptionYear"],
+            # "year": ["inceptionYear"],  # not supported by somesy + does not really change
+            # "project_slug": ["artifactId"],  # not supported by somesy for sync
             "license": ["licenses", "license"],
             "homepage": ["url"],
-            "project_slug": ["artifactId"],
+            "repository": ["scm"],
+            "documentation": ["distributionManagement", "site"],
             "authors": ["developers", "developer"],
             "contributors": ["contributors", "contributor"],
         }
@@ -45,10 +46,11 @@ def _init_new_file(self):
         """Initialize new pom.xml file."""
         pom = XMLProxy(ET.Element("project", POM_ROOT_ATRS))
         pom["properties"] = {"info.versionScheme": "semver-spec"}
-        pom.write(self.path, default_namespace=POM_URL)
+        pom.write(self.path)
 
     def _load(self):
         """Load the POM file."""
+        ET.register_namespace("", POM_URL)  # register POM as default xml namespace
         self._data = XMLProxy.parse(self.path, default_namespace=POM_URL)
 
     def _validate(self):
@@ -57,30 +59,158 @@ def _validate(self):
 
     def save(self, path: Optional[Path] = None) -> None:
         """Save the POM DOM to a file."""
-        path = path or self.path
-        self._data.write(path)
+        self._data.write(path or self.path, default_namespace=None)
+
+    def _get_property(
+        self,
+        key: Union[str, List[str]],
+        *,
+        only_first: bool = False,
+        remove: bool = False,
+    ) -> Optional[Any]:
+        elem = super()._get_property(key, only_first=only_first, remove=remove)
+        if elem is not None:
+            if isinstance(elem, list):
+                return [e.to_jsonlike() for e in elem]
+            else:
+                return elem.to_jsonlike()
+        return None
 
     @staticmethod
     def _from_person(person: Person):
-        """Convert project metadata person object to cff dict for person format."""
-        ret = {}
-        person_id = person.orcid or person.to_name_email_string()
+        """Convert person object to dict for POM XML person format."""
+        ret: Dict[str, Any] = {}
+        person_id = str(person.orcid) or person.to_name_email_string()
         ret["id"] = person_id
-        ret["name"] = person.name
+        ret["name"] = person.full_name
         ret["email"] = person.email
         if person.orcid:
-            ret["url"] = person.orcid
+            ret["url"] = str(person.orcid)
         if person.contribution_types:
-            ret["roles"] = dict(role=person.contribution_types)
+            ret["roles"] = dict(role=[c.value for c in person.contribution_types])
         return ret
 
     @staticmethod
     def _to_person(person_obj) -> Person:
-        """Parse CFF Person to a somesy Person."""
+        """Parse POM XML person to a somesy Person."""
+        print(person_obj)
+        names = person_obj["name"].split()
+        gnames = " ".join(names[:-1])
+        fname = names[-1]
+        email = person_obj["email"]
+        url = person_obj.get("url")
+        maybe_orcid = url if url.find("orcid.org") >= 0 else None
+        if roles := person_obj.get("roles"):
+            contr = roles["role"]
+        else:
+            contr = None
+
         return Person(
-            name=person_obj["name"],
-            email=person_obj["email"],
-            orcid=person_obj["orcid"],
-            contribution_types=person_obj["roles"]["role"],
+            given_names=gnames,
+            family_names=fname,
+            email=email,
+            orcid=maybe_orcid,
+            contribution_types=contr,
         )
-        raise NotImplementedError
+
+    # no search keywords supported in POM
+    @property
+    def keywords(self) -> Optional[List[str]]:
+        """Return the keywords of the project."""
+        pass
+
+    @keywords.setter
+    def keywords(self, keywords: List[str]) -> None:
+        """Set the keywords of the project."""
+        pass
+
+    # authors must be a list
+    @property
+    def authors(self):
+        """Return the authors of the project."""
+        authors = self._get_property(self._get_key("authors"))
+        return authors if isinstance(authors, list) else [authors]
+
+    @authors.setter
+    def authors(self, authors: List[Person]) -> None:
+        """Set the authors of the project."""
+        authors = [self._from_person(c) for c in authors]
+        self._set_property(self._get_key("authors"), authors)
+
+    # contributors must be a list
+    @property
+    def contributors(self):
+        """Return the contributors of the project."""
+        contr = self._get_property(self._get_key("contributors"))
+        if contr is None:
+            return []
+        return contr if isinstance(contr, list) else [contr]
+
+    @contributors.setter
+    def contributors(self, contributors: List[Person]) -> None:
+        """Set the contributors of the project."""
+        contr = [self._from_person(c) for c in contributors]
+        self._set_property(self._get_key("contributors"), contr)
+
+    # no maintainers supported im POM, only developers and contributors
+    @property
+    def maintainers(self):
+        """Return the maintainers of the project."""
+        return []
+
+    @maintainers.setter
+    def maintainers(self, maintainers: List[Person]) -> None:
+        """Set the maintainers of the project."""
+        pass
+
+    # only one project license supported in somesy (POM can have many)
+    @property
+    def license(self) -> Optional[str]:
+        """Return the license of the project."""
+        lic = self._get_property(self._get_key("license"), only_first=True)
+        return lic.get("name") if lic is not None else None
+
+    @license.setter
+    def license(self, license: Optional[str]) -> None:
+        """Set the license of the project."""
+        self._set_property(
+            self._get_key("license"), dict(name=license, distribution="repo")
+        )
+
+    @property
+    def repository(self) -> Optional[Union[str, dict]]:
+        """Return the repository url of the project."""
+        repo = super().repository
+        if isinstance(repo, str):
+            return repo
+        return repo.get("url") if repo is not None else None
+
+    @repository.setter
+    def repository(self, value: Optional[Union[str, dict]]) -> None:
+        """Set the repository url of the project."""
+        self._set_property(
+            self._get_key("repository"), dict(name="git repository", url=value)
+        )
+
+    @property
+    def documentation(self) -> Optional[Union[str, dict]]:
+        """Return the documentation url of the project."""
+        docs = super().documentation
+        if isinstance(docs, str):
+            return docs
+        return docs.get("url") if docs is not None else None
+
+    @documentation.setter
+    def documentation(self, value: Optional[Union[str, dict]]) -> None:
+        """Set the documentation url of the project."""
+        self._set_property(
+            self._get_key("documentation"), dict(name="documentation site", url=value)
+        )
+
+    def sync(self, metadata) -> None:
+        """Sync codemeta.json with project metadata.
+
+        Use existing sync function from ProjectMetadataWriter but update repository and contributors.
+        """
+        super().sync(metadata)
+        self.contributors = self._sync_person_list(self.contributors, metadata.people)