Coverage for src/somesy/core/writer.py: 97%
230 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-04 11:35 +0000
1"""Project metadata writer base-class."""
3import logging
4from abc import ABC, abstractmethod
5from collections.abc import Sequence
6from pathlib import Path
7from typing import Any
9from somesy.core.models import Entity, Person, ProjectMetadata
11logger = logging.getLogger("somesy")
14class IgnoreKey:
15 """Special marker to be passed for dropping a key from serialization."""
18FieldKeyMapping = dict[str, str | list[str] | IgnoreKey]
19"""Type to be used for the dict passed as `direct_mappings`."""
21DictLike = Any
22"""Dict-like that supports getitem, setitem, delitem, etc.
24NOTE: This should be probably turned into a proper protocol.
25"""
28class ProjectMetadataWriter(ABC):
29 """Base class for Project Metadata Output Wrapper.
31 All supported output formats are implemented as subclasses.
32 """
34 def __init__(
35 self,
36 path: Path,
37 *,
38 create_if_not_exists: bool | None = False,
39 direct_mappings: FieldKeyMapping | None = None,
40 merge: bool | None = False,
41 pass_validation: bool | None = False,
42 ) -> None:
43 """Initialize the Project Metadata Output Wrapper.
45 Use the `direct_mappings` dict to define
46 format-specific location for certain fields,
47 if no additional processing is needed that
48 requires a customized setter.
50 Args:
51 path: Path to target output file.
52 create_if_not_exists: Create an empty CFF file if not exists. Defaults to True.
53 direct_mappings: Dict with direct mappings of keys between somesy and target
54 merge: Merge the output file with an existing file. Defaults to False.
55 pass_validation: Pass validation for all output files. Defaults to False.
57 """
58 self._data: DictLike = {}
59 self.path = path if isinstance(path, Path) else Path(path)
60 self.create_if_not_exists = create_if_not_exists
61 self.direct_mappings = direct_mappings or {}
62 self.merge = merge
63 self.pass_validation = pass_validation
64 if self.path.is_file():
65 self._load()
66 if not self.pass_validation:
67 self._validate()
68 else:
69 if self.create_if_not_exists:
70 self._init_new_file()
71 self._load()
72 else:
73 raise FileNotFoundError(f"The file {self.path} does not exist.")
75 def _init_new_file(self) -> None:
76 """Create an new suitable target file.
78 Override to initialize file with minimal contents, if needed.
79 Make sure to set `self._data` to match the contents.
80 """
81 self.path.touch()
83 @abstractmethod
84 def _load(self):
85 """Load the output file and validate it.
87 Implement this method so that it loads the file `self.path`
88 into the `self._data` dict.
90 The file is guaranteed to exist.
91 """
93 @abstractmethod
94 def _validate(self) -> None:
95 """Validate the target file data.
97 Implement this method so that it checks
98 the validity of the metadata (relevant to somesy)
99 in that file and raises exceptions on failure.
100 """
102 @abstractmethod
103 def save(self, path: Path | None) -> None:
104 """Save the output file to the given path.
106 Implement this in a way that will carefully
107 update the target file with new metadata
108 without destroying its other contents or structure.
109 """
111 def _get_property(
112 self,
113 key: str | list[str] | IgnoreKey,
114 *,
115 only_first: bool = False,
116 remove: bool = False,
117 ) -> Any:
118 """Get a property from the data.
120 Override this to e.g. rewrite the retrieved key
121 (e.g. if everything relevant is in some subobject).
123 Args:
124 key: Name of the key or sequence of multiple keys to retrieve the value.
125 only_first: If True, returns only first entry if the value is a list.
126 remove: If True, will remove the retrieved value and clean up the dict.
128 """
129 if isinstance(key, IgnoreKey):
130 return None
131 key_path = [key] if isinstance(key, str) else key
133 curr: Any = self._data
134 seq = [curr]
135 for k in key_path:
136 curr = curr.get(k)
137 curr = curr[0] if isinstance(curr, list) and only_first else curr
138 seq.append(curr)
139 if curr is None:
140 return None
142 if remove:
143 seq.pop()
144 del seq[-1][key_path[-1]] # remove leaf value
145 # clean up the tree
146 for path_key, dct in reversed(
147 list(zip(key_path[:-1], seq[:-1], strict=False))
148 ):
149 if not dct.get(path_key):
150 del dct[path_key]
152 if isinstance(curr, list) and only_first:
153 return curr[0]
154 return curr
156 def _set_property(self, key: str | list[str] | IgnoreKey, value: Any) -> None:
157 """Set a property in the data.
159 Note if there are lists along the path, they are cleared out.
161 Override this to e.g. rewrite the retrieved key
162 (e.g. if everything relevant is in some subobject).
163 """
164 if isinstance(key, IgnoreKey):
165 return
166 key_path = [key] if isinstance(key, str) else key
168 if not value: # remove value and clean up the sub-dict
169 self._get_property(key_path, remove=True)
170 return
172 # create path on the fly if needed
173 curr = self._data
174 for path_key in key_path[:-1]:
175 if path_key not in curr:
176 curr[path_key] = {}
177 curr = curr[path_key]
179 curr[key_path[-1]] = value
181 # ----
182 # special handling for person metadata
184 def _merge_person_metadata(
185 self,
186 old: Sequence[Person | Entity],
187 new: Sequence[Person | Entity],
188 ) -> list[Person | Entity]:
189 """Update metadata of a list of persons.
191 Will identify people based on orcid, email or full name.
193 If old list has same person listed multiple times,
194 the resulting list will too (we cannot correctly merge for external formats.)
195 """
196 new_people = [] # list for new people (e.g. added authors)
197 # flag, meaning "person was not removed"
198 still_exists = [False for i in range(len(old))]
199 # copies of old person data, to be modified
200 modified_people = [p.model_copy() for p in old]
202 # try to match new people to existing old ones
203 # (inefficient, but author list are not that long usually)
204 for person_meta in new:
205 person_update = person_meta.model_dump()
206 person_existed = False
207 for i in range(len(modified_people)):
208 person = modified_people[i]
209 if not person.same_person(person_meta):
210 continue
212 # not new person (-> will not append new record)
213 person_existed = True
214 # still exists (-> will not be removed from list)
215 still_exists[i] = True
217 # if there were changes -> update person
218 overlapping_fields = person.model_dump(
219 include=set(person_update.keys())
220 )
221 if person_update != overlapping_fields:
222 modified_people[i] = person.model_copy(update=person_update)
224 # show effective update in debug log
225 old_fmt = self._from_person(person)
226 new_fmt = self._from_person(modified_people[i])
227 if old_fmt != new_fmt:
228 logger.debug(f"Updating person\n{old_fmt}\nto\n{new_fmt}")
230 if not person_existed:
231 new_people.append(person_meta)
233 # show added and removed people in debug log
234 removed_people = [old[i] for i in range(len(old)) if not still_exists[i]]
235 for person in removed_people:
236 logger.debug(f"Removing person\n{self._from_person(person)}")
237 for person in new_people:
238 logger.debug(f"Adding person\n{self._from_person(person)}")
240 # return updated list of (still existing) people,
241 # and all new people coming after them.
242 existing_modified = [
243 modified_people[i] for i in range(len(old)) if still_exists[i]
244 ]
245 return existing_modified + new_people
247 def _sync_person_list(
248 self, old: list[Any], new: Sequence[Person | Entity]
249 ) -> list[Any]:
250 """Sync a list of persons with new metadata.
252 Args:
253 old (List[Any]): list of persons in format-specific representation
254 new (List[Person]): list of persons in somesy representation
256 Returns:
257 List[Any]: updated list of persons in format-specific representation
259 """
260 old_people: list[Person | Entity] = self._parse_people(old)
262 # check if people are unique
263 def filter_unique(
264 people: Sequence[Person | Entity],
265 ) -> list[Person | Entity]:
266 """Filter out duplicate people from a list."""
267 if people is None or len(people) == 0:
268 return []
270 unique_people: list[Person | Entity] = []
271 # use same_person method to check if people are unique
272 for person in people:
273 if not any(person.same_person(p) for p in unique_people):
274 unique_people.append(person)
276 return unique_people
278 old_people_unique = filter_unique(old_people)
279 new_people_unique = filter_unique(new)
281 return self._merge_person_metadata(old_people_unique, new_people_unique)
283 def _sync_authors(self, metadata: ProjectMetadata) -> None:
284 """Sync output file authors with authors from metadata.
286 This method is existing for the publication_author special case
287 when synchronizing to CITATION.cff.
288 """
289 if self.authors is None or len(self.authors) == 0:
290 self.authors = metadata.authors()
291 else:
292 self.authors = self._sync_person_list(self.authors, metadata.authors())
294 def sync(self, metadata: ProjectMetadata) -> None:
295 """Sync output file with other metadata files."""
296 self.name = metadata.name
297 self.description = metadata.description
299 if metadata.version:
300 self.version = metadata.version
302 if metadata.keywords:
303 self.keywords = metadata.keywords
305 self._sync_authors(metadata)
306 self.maintainers = self._sync_person_list(
307 self.maintainers, metadata.maintainers()
308 )
310 licenses = metadata.license
311 self.license = (
312 licenses[0].value if isinstance(licenses, list) else licenses.value
313 )
315 self.homepage = str(metadata.homepage) if metadata.homepage else None
316 self.repository = str(metadata.repository) if metadata.repository else None
317 self.documentation = (
318 str(metadata.documentation) if metadata.documentation else None
319 )
321 def harvest_metadata(self) -> dict[str, Any]:
322 """Return metadata read from this endpoint in Somesy model terms."""
323 data: dict[str, Any] = {}
324 for field in (
325 "name",
326 "version",
327 "description",
328 "license",
329 "homepage",
330 "repository",
331 "documentation",
332 "keywords",
333 ):
334 try:
335 value = getattr(self, field)
336 except (KeyError, TypeError):
337 continue
338 if value not in (None, [], ""):
339 data[field] = value
341 people: list[Person] = []
342 entities: list[Entity] = []
343 for role in ("authors", "maintainers", "contributors"):
344 for person in self._parse_people(getattr(self, role) or []):
345 updates = {"author": True} if role == "authors" else {}
346 updates.update({"maintainer": True} if role == "maintainers" else {})
347 person = person.model_copy(update=updates)
348 if isinstance(person, Entity):
349 entities.append(person)
350 else:
351 people.append(person)
353 if people:
354 data["people"] = people
355 if entities:
356 data["entities"] = entities
357 return data
359 @staticmethod
360 @abstractmethod
361 def _from_person(person: Person | Entity) -> Any:
362 """Convert a `Person` or `Entity` object into suitable target format."""
364 @staticmethod
365 @abstractmethod
366 def _to_person(person_obj: Any) -> Person | Entity | None:
367 """Convert an object representing a person into a `Person` or `Entity` object."""
369 @classmethod
370 def _parse_people(cls, people: list[Any] | None) -> list[Person | Entity]:
371 """Return a list of Persons and Entities parsed from list of format-specific people representations."""
372 # remove None values
373 return [
374 person
375 for p in people or []
376 if p is not None
377 if (person := cls._to_person(p)) is not None
378 ]
380 # ----
381 # individual magic getters and setters
383 def _get_key(self, key: str) -> str | list[str] | IgnoreKey:
384 """Get a key itself."""
385 return self.direct_mappings.get(key) or key
387 @property
388 def name(self):
389 """Return the name of the project."""
390 return self._get_property(self._get_key("name"))
392 @name.setter
393 def name(self, name: str) -> None:
394 """Set the name of the project."""
395 self._set_property(self._get_key("name"), name)
397 @property
398 def version(self) -> str | None:
399 """Return the version of the project."""
400 return self._get_property(self._get_key("version"))
402 @version.setter
403 def version(self, version: str | None) -> None:
404 """Set the version of the project."""
405 self._set_property(self._get_key("version"), version)
407 @property
408 def description(self) -> str | None:
409 """Return the description of the project."""
410 return self._get_property(self._get_key("description"))
412 @description.setter
413 def description(self, description: str) -> None:
414 """Set the description of the project."""
415 self._set_property(self._get_key("description"), description)
417 @property
418 def authors(self):
419 """Return the authors of the project."""
420 authors = self._get_property(self._get_key("authors"))
421 if authors is None or len(authors) == 0:
422 return []
424 # only return authors that can be converted to Person
425 authors_validated = [
426 author for author in authors if self._to_person(author) is not None
427 ]
428 return authors_validated
430 @authors.setter
431 def authors(self, authors: list[Person | Entity]) -> None:
432 """Set the authors of the project."""
433 authors = [self._from_person(c) for c in authors]
434 self._set_property(self._get_key("authors"), authors)
436 @property
437 def maintainers(self):
438 """Return the maintainers of the project."""
439 maintainers = self._get_property(self._get_key("maintainers"))
440 if maintainers is None:
441 return []
443 # only return maintainers that can be converted to Person
444 maintainers_validated = [
445 maintainer
446 for maintainer in maintainers
447 if self._to_person(maintainer) is not None
448 ]
449 return maintainers_validated
451 @maintainers.setter
452 def maintainers(self, maintainers: list[Person | Entity]) -> None:
453 """Set the maintainers of the project."""
454 maintainers = [self._from_person(c) for c in maintainers]
455 self._set_property(self._get_key("maintainers"), maintainers)
457 @property
458 def contributors(self):
459 """Return the contributors of the project."""
460 return self._get_property(self._get_key("contributors"))
462 @contributors.setter
463 def contributors(self, contributors: list[Person | Entity]) -> None:
464 """Set the contributors of the project."""
465 contributors = [self._from_person(c) for c in contributors]
466 self._set_property(self._get_key("contributors"), contributors)
468 @property
469 def keywords(self) -> list[str] | None:
470 """Return the keywords of the project."""
471 return self._get_property(self._get_key("keywords"))
473 @keywords.setter
474 def keywords(self, keywords: list[str]) -> None:
475 """Set the keywords of the project."""
476 self._set_property(self._get_key("keywords"), keywords)
478 @property
479 def license(self) -> Any:
480 """Return the license of the project."""
481 return self._get_property(self._get_key("license"))
483 @license.setter
484 def license(self, license: Any) -> None:
485 """Set the license of the project."""
486 self._set_property(self._get_key("license"), license)
488 @property
489 def homepage(self) -> str | None:
490 """Return the homepage url of the project."""
491 return self._get_property(self._get_key("homepage"))
493 @homepage.setter
494 def homepage(self, value: str | None) -> None:
495 """Set the homepage url of the project."""
496 self._set_property(self._get_key("homepage"), value)
498 @property
499 def repository(self) -> str | dict | None:
500 """Return the repository url of the project."""
501 return self._get_property(self._get_key("repository"))
503 @repository.setter
504 def repository(self, value: str | dict | None) -> None:
505 """Set the repository url of the project."""
506 self._set_property(self._get_key("repository"), value)
508 @property
509 def documentation(self) -> str | dict | None:
510 """Return the documentation url of the project."""
511 return self._get_property(self._get_key("documentation"))
513 @documentation.setter
514 def documentation(self, value: str | dict | None) -> None:
515 """Set the documentation url of the project."""
516 self._set_property(self._get_key("documentation"), value)