Coverage for src/somesy/core/writer.py: 97%
233 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 11:27 +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 if metadata.name is not None:
297 self.name = metadata.name
298 if metadata.description is not None:
299 self.description = metadata.description
301 if metadata.version:
302 self.version = metadata.version
304 if metadata.keywords:
305 self.keywords = metadata.keywords
307 if metadata.authors():
308 self._sync_authors(metadata)
309 self.maintainers = self._sync_person_list(
310 self.maintainers, metadata.maintainers()
311 )
313 if licenses := metadata.license:
314 self.license = (
315 licenses[0].value if isinstance(licenses, list) else licenses.value
316 )
318 self.homepage = str(metadata.homepage) if metadata.homepage else None
319 self.repository = str(metadata.repository) if metadata.repository else None
320 self.documentation = (
321 str(metadata.documentation) if metadata.documentation else None
322 )
324 def harvest_metadata(self) -> dict[str, Any]:
325 """Return metadata read from this endpoint in Somesy model terms."""
326 data: dict[str, Any] = {}
327 for field in (
328 "name",
329 "version",
330 "description",
331 "license",
332 "homepage",
333 "repository",
334 "documentation",
335 "keywords",
336 ):
337 try:
338 value = getattr(self, field)
339 except (KeyError, TypeError):
340 continue
341 if value not in (None, [], ""):
342 data[field] = value
344 people: list[Person] = []
345 entities: list[Entity] = []
346 for role in ("authors", "maintainers", "contributors"):
347 for person in self._parse_people(getattr(self, role) or []):
348 updates = {"author": True} if role == "authors" else {}
349 updates.update({"maintainer": True} if role == "maintainers" else {})
350 person = person.model_copy(update=updates)
351 if isinstance(person, Entity):
352 entities.append(person)
353 else:
354 people.append(person)
356 if people:
357 data["people"] = people
358 if entities:
359 data["entities"] = entities
360 return data
362 @staticmethod
363 @abstractmethod
364 def _from_person(person: Person | Entity) -> Any:
365 """Convert a `Person` or `Entity` object into suitable target format."""
367 @staticmethod
368 @abstractmethod
369 def _to_person(person_obj: Any) -> Person | Entity | None:
370 """Convert an object representing a person into a `Person` or `Entity` object."""
372 @classmethod
373 def _parse_people(cls, people: list[Any] | None) -> list[Person | Entity]:
374 """Return a list of Persons and Entities parsed from list of format-specific people representations."""
375 # remove None values
376 return [
377 person
378 for p in people or []
379 if p is not None
380 if (person := cls._to_person(p)) is not None
381 ]
383 # ----
384 # individual magic getters and setters
386 def _get_key(self, key: str) -> str | list[str] | IgnoreKey:
387 """Get a key itself."""
388 return self.direct_mappings.get(key) or key
390 @property
391 def name(self):
392 """Return the name of the project."""
393 return self._get_property(self._get_key("name"))
395 @name.setter
396 def name(self, name: str) -> None:
397 """Set the name of the project."""
398 self._set_property(self._get_key("name"), name)
400 @property
401 def version(self) -> str | None:
402 """Return the version of the project."""
403 return self._get_property(self._get_key("version"))
405 @version.setter
406 def version(self, version: str | None) -> None:
407 """Set the version of the project."""
408 self._set_property(self._get_key("version"), version)
410 @property
411 def description(self) -> str | None:
412 """Return the description of the project."""
413 return self._get_property(self._get_key("description"))
415 @description.setter
416 def description(self, description: str) -> None:
417 """Set the description of the project."""
418 self._set_property(self._get_key("description"), description)
420 @property
421 def authors(self):
422 """Return the authors of the project."""
423 authors = self._get_property(self._get_key("authors"))
424 if authors is None or len(authors) == 0:
425 return []
427 # only return authors that can be converted to Person
428 authors_validated = [
429 author for author in authors if self._to_person(author) is not None
430 ]
431 return authors_validated
433 @authors.setter
434 def authors(self, authors: list[Person | Entity]) -> None:
435 """Set the authors of the project."""
436 authors = [self._from_person(c) for c in authors]
437 self._set_property(self._get_key("authors"), authors)
439 @property
440 def maintainers(self):
441 """Return the maintainers of the project."""
442 maintainers = self._get_property(self._get_key("maintainers"))
443 if maintainers is None:
444 return []
446 # only return maintainers that can be converted to Person
447 maintainers_validated = [
448 maintainer
449 for maintainer in maintainers
450 if self._to_person(maintainer) is not None
451 ]
452 return maintainers_validated
454 @maintainers.setter
455 def maintainers(self, maintainers: list[Person | Entity]) -> None:
456 """Set the maintainers of the project."""
457 maintainers = [self._from_person(c) for c in maintainers]
458 self._set_property(self._get_key("maintainers"), maintainers)
460 @property
461 def contributors(self):
462 """Return the contributors of the project."""
463 return self._get_property(self._get_key("contributors"))
465 @contributors.setter
466 def contributors(self, contributors: list[Person | Entity]) -> None:
467 """Set the contributors of the project."""
468 contributors = [self._from_person(c) for c in contributors]
469 self._set_property(self._get_key("contributors"), contributors)
471 @property
472 def keywords(self) -> list[str] | None:
473 """Return the keywords of the project."""
474 return self._get_property(self._get_key("keywords"))
476 @keywords.setter
477 def keywords(self, keywords: list[str]) -> None:
478 """Set the keywords of the project."""
479 self._set_property(self._get_key("keywords"), keywords)
481 @property
482 def license(self) -> Any:
483 """Return the license of the project."""
484 return self._get_property(self._get_key("license"))
486 @license.setter
487 def license(self, license: Any) -> None:
488 """Set the license of the project."""
489 self._set_property(self._get_key("license"), license)
491 @property
492 def homepage(self) -> str | None:
493 """Return the homepage url of the project."""
494 return self._get_property(self._get_key("homepage"))
496 @homepage.setter
497 def homepage(self, value: str | None) -> None:
498 """Set the homepage url of the project."""
499 self._set_property(self._get_key("homepage"), value)
501 @property
502 def repository(self) -> str | dict | None:
503 """Return the repository url of the project."""
504 return self._get_property(self._get_key("repository"))
506 @repository.setter
507 def repository(self, value: str | dict | None) -> None:
508 """Set the repository url of the project."""
509 self._set_property(self._get_key("repository"), value)
511 @property
512 def documentation(self) -> str | dict | None:
513 """Return the documentation url of the project."""
514 return self._get_property(self._get_key("documentation"))
516 @documentation.setter
517 def documentation(self, value: str | dict | None) -> None:
518 """Set the documentation url of the project."""
519 self._set_property(self._get_key("documentation"), value)