Coverage for src/somesy/core/writer.py: 95%
258 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-11-03 13:31 +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
9import tomlkit
10from tomlkit.items import AoT, Array, InlineTable
12from somesy.core.models import Entity, Person, ProjectMetadata
14logger = logging.getLogger("somesy")
17def toml_array(value: list) -> Array:
18 """Build a multiline tomlkit array, with spaced inline tables."""
19 array = tomlkit.array()
20 array.extend(value)
21 array.multiline(True)
22 # Ensure whitespace after commas in inline tables
23 for item in array:
24 if isinstance(item, InlineTable):
25 # Rebuild the inline table with desired formatting
26 formatted_item = tomlkit.inline_table()
27 for k, v in item.value.items():
28 formatted_item[k] = v
29 formatted_item.trivia.trail = " " # Add space after each comma
30 array[array.index(item)] = formatted_item
31 return array
34def multiline_description(table: Any) -> None:
35 """Store a description containing line breaks as a multiline TOML string."""
36 if "description" in table and "\n" in table["description"]:
37 table["description"] = tomlkit.string(table["description"], multiline=True)
40def format_toml_data(data: Any) -> None:
41 """Prepare a tomlkit document for saving: multiline description, multiline arrays."""
42 multiline_description(data)
44 # Handle arrays with proper formatting (snapshot, since we assign while iterating)
45 for key, value in list(data.items()):
46 if isinstance(value, list) and not isinstance(value, AoT):
47 data[key] = toml_array(value)
48 else:
49 data[key] = value
52class IgnoreKey:
53 """Special marker to be passed for dropping a key from serialization."""
56FieldKeyMapping = dict[str, str | list[str] | IgnoreKey]
57"""Type to be used for the dict passed as `direct_mappings`."""
59DictLike = Any
60"""Dict-like that supports getitem, setitem, delitem, etc.
62NOTE: This should be probably turned into a proper protocol.
63"""
66def license_expression(licenses) -> str:
67 """Convert one or more license identifiers to an SPDX expression."""
68 return " OR ".join(
69 str(license)
70 for license in (licenses if isinstance(licenses, list) else [licenses])
71 )
74class ProjectMetadataWriter(ABC):
75 """Base class for Project Metadata Output Wrapper.
77 All supported output formats are implemented as subclasses.
78 """
80 def __init__(
81 self,
82 path: Path,
83 *,
84 create_if_not_exists: bool | None = False,
85 direct_mappings: FieldKeyMapping | None = None,
86 merge: bool | None = False,
87 pass_validation: bool | None = False,
88 ) -> None:
89 """Initialize the Project Metadata Output Wrapper.
91 Use the `direct_mappings` dict to define
92 format-specific location for certain fields,
93 if no additional processing is needed that
94 requires a customized setter.
96 Args:
97 path: Path to target output file.
98 create_if_not_exists: Create an empty file if not exists. Defaults to False.
99 direct_mappings: Dict with direct mappings of keys between somesy and target
100 merge: Merge the output file with an existing file. Defaults to False.
101 pass_validation: Pass validation for all output files. Defaults to False.
103 """
104 self._data: DictLike = {}
105 self.path = path if isinstance(path, Path) else Path(path)
106 self.create_if_not_exists = create_if_not_exists
107 self.direct_mappings = direct_mappings or {}
108 self.merge = merge
109 self.pass_validation = pass_validation
110 if self.path.is_file():
111 self._load()
112 if not self.pass_validation:
113 self._validate()
114 else:
115 if self.create_if_not_exists:
116 self._init_new_file()
117 self._load()
118 else:
119 raise FileNotFoundError(f"The file {self.path} does not exist.")
121 def _init_new_file(self) -> None:
122 """Create an new suitable target file.
124 Override to initialize file with minimal contents, if needed.
125 Make sure to set `self._data` to match the contents.
126 """
127 self.path.touch()
129 @abstractmethod
130 def _load(self):
131 """Load the output file and validate it.
133 Implement this method so that it loads the file `self.path`
134 into the `self._data` dict.
136 The file is guaranteed to exist.
137 """
139 @abstractmethod
140 def _validate(self) -> None:
141 """Validate the target file data.
143 Implement this method so that it checks
144 the validity of the metadata (relevant to somesy)
145 in that file and raises exceptions on failure.
146 """
148 @abstractmethod
149 def save(self, path: Path | None) -> None:
150 """Save the output file to the given path.
152 Implement this in a way that will carefully
153 update the target file with new metadata
154 without destroying its other contents or structure.
155 """
157 def _get_property(
158 self,
159 key: str | list[str] | IgnoreKey,
160 *,
161 remove: bool = False,
162 ) -> Any:
163 """Get a property from the data.
165 Override this to e.g. rewrite the retrieved key
166 (e.g. if everything relevant is in some subobject).
168 Args:
169 key: Name of the key or sequence of multiple keys to retrieve the value.
170 remove: If True, will remove the retrieved value and clean up the dict.
172 """
173 if isinstance(key, IgnoreKey):
174 return None
175 key_path = [key] if isinstance(key, str) else key
177 curr: Any = self._data
178 seq = [curr]
179 for k in key_path:
180 curr = curr.get(k)
181 seq.append(curr)
182 if curr is None:
183 return None
185 if remove:
186 seq.pop()
187 del seq[-1][key_path[-1]] # remove leaf value
188 # clean up the tree
189 for path_key, dct in reversed(
190 list(zip(key_path[:-1], seq[:-1], strict=False))
191 ):
192 if not dct.get(path_key):
193 del dct[path_key]
195 return curr
197 def _set_property(self, key: str | list[str] | IgnoreKey, value: Any) -> None:
198 """Set a property in the data.
200 Note if there are lists along the path, they are cleared out.
202 Override this to e.g. rewrite the retrieved key
203 (e.g. if everything relevant is in some subobject).
204 """
205 if isinstance(key, IgnoreKey):
206 return
207 key_path = [key] if isinstance(key, str) else key
209 if not value: # remove value and clean up the sub-dict
210 self._get_property(key_path, remove=True)
211 return
213 # create path on the fly if needed
214 curr = self._data
215 for path_key in key_path[:-1]:
216 if path_key not in curr:
217 curr[path_key] = {}
218 curr = curr[path_key]
220 curr[key_path[-1]] = value
222 # ----
223 # special handling for person metadata
225 def _merge_person_metadata(
226 self,
227 old: Sequence[Person | Entity],
228 new: Sequence[Person | Entity],
229 ) -> list[Person | Entity]:
230 """Update metadata of a list of persons.
232 Will identify people based on orcid, email or full name.
234 If old list has same person listed multiple times,
235 the resulting list will too (we cannot correctly merge for external formats.)
236 """
237 new_people = [] # list for new people (e.g. added authors)
238 # flag, meaning "person was not removed"
239 still_exists = [False for i in range(len(old))]
240 # copies of old person data, to be modified
241 modified_people = [p.model_copy() for p in old]
243 # try to match new people to existing old ones
244 # (inefficient, but author list are not that long usually)
245 for person_meta in new:
246 person_update = person_meta.model_dump()
247 person_existed = False
248 for i in range(len(modified_people)):
249 person = modified_people[i]
250 if not person.same_person(person_meta):
251 continue
253 # not new person (-> will not append new record)
254 person_existed = True
255 # still exists (-> will not be removed from list)
256 still_exists[i] = True
258 # if there were changes -> update person
259 overlapping_fields = person.model_dump(
260 include=set(person_update.keys())
261 )
262 if person_update != overlapping_fields:
263 modified_people[i] = person.model_copy(update=person_update)
265 # show effective update in debug log
266 old_fmt = self._from_person(person)
267 new_fmt = self._from_person(modified_people[i])
268 if old_fmt != new_fmt:
269 logger.debug(f"Updating person\n{old_fmt}\nto\n{new_fmt}")
271 if not person_existed:
272 new_people.append(person_meta)
274 # show added and removed people in debug log
275 removed_people = [old[i] for i in range(len(old)) if not still_exists[i]]
276 for person in removed_people:
277 logger.debug(f"Removing person\n{self._from_person(person)}")
278 for person in new_people:
279 logger.debug(f"Adding person\n{self._from_person(person)}")
281 # return updated list of (still existing) people,
282 # and all new people coming after them.
283 existing_modified = [
284 modified_people[i] for i in range(len(old)) if still_exists[i]
285 ]
286 return existing_modified + new_people
288 def _sync_person_list(
289 self, old: list[Any], new: Sequence[Person | Entity]
290 ) -> list[Any]:
291 """Sync a list of persons with new metadata.
293 Args:
294 old (List[Any]): list of persons in format-specific representation
295 new (List[Person]): list of persons in somesy representation
297 Returns:
298 List[Any]: updated list of persons in format-specific representation
300 """
301 old_people: list[Person | Entity] = self._parse_people(old)
303 # check if people are unique
304 def filter_unique(
305 people: Sequence[Person | Entity],
306 ) -> list[Person | Entity]:
307 """Filter out duplicate people from a list."""
308 if people is None or len(people) == 0:
309 return []
311 unique_people: list[Person | Entity] = []
312 # use same_person method to check if people are unique
313 for person in people:
314 if not any(person.same_person(p) for p in unique_people):
315 unique_people.append(person)
317 return unique_people
319 old_people_unique = filter_unique(old_people)
320 new_people_unique = filter_unique(new)
322 return self._merge_person_metadata(old_people_unique, new_people_unique)
324 def _sync_authors(self, metadata: ProjectMetadata) -> None:
325 """Sync output file authors with authors from metadata.
327 This method is existing for the publication_author special case
328 when synchronizing to CITATION.cff.
329 """
330 if self.authors is None or len(self.authors) == 0:
331 self.authors = metadata.authors()
332 else:
333 self.authors = self._sync_person_list(self.authors, metadata.authors())
335 def sync(self, metadata: ProjectMetadata) -> None:
336 """Sync output file with other metadata files."""
337 if metadata.name is not None:
338 self.name = metadata.name
339 if metadata.description is not None:
340 self.description = metadata.description
342 if metadata.version:
343 self.version = metadata.version
345 if metadata.keywords:
346 self.keywords = metadata.keywords
348 if metadata.authors():
349 self._sync_authors(metadata)
350 self.maintainers = self._sync_person_list(
351 self.maintainers, metadata.maintainers()
352 )
354 if licenses := metadata.license:
355 self.license = license_expression(licenses)
357 # keep existing values in the target file if somesy has none
358 if metadata.homepage:
359 self.homepage = str(metadata.homepage)
360 if metadata.repository:
361 self.repository = str(metadata.repository)
362 if metadata.documentation:
363 self.documentation = str(metadata.documentation)
365 def harvest_metadata(self) -> dict[str, Any]:
366 """Return metadata read from this endpoint in Somesy model terms."""
367 data: dict[str, Any] = {}
368 for field in (
369 "name",
370 "version",
371 "description",
372 "license",
373 "homepage",
374 "repository",
375 "documentation",
376 "keywords",
377 ):
378 try:
379 value = getattr(self, field)
380 except (KeyError, TypeError):
381 continue
382 if value not in (None, [], ""):
383 data[field] = value
385 people: list[Person] = []
386 entities: list[Entity] = []
387 for role in ("authors", "maintainers", "contributors"):
388 for person in self._parse_people(getattr(self, role) or []):
389 updates = (
390 {"author": True, "publication_author": True}
391 if role == "authors"
392 else {}
393 )
394 updates.update({"maintainer": True} if role == "maintainers" else {})
395 person = person.model_copy(update=updates)
396 if isinstance(person, Entity):
397 entities.append(person)
398 else:
399 people.append(person)
401 if people:
402 data["people"] = people
403 if entities:
404 data["entities"] = entities
405 return data
407 @staticmethod
408 @abstractmethod
409 def _from_person(person: Person | Entity) -> Any:
410 """Convert a `Person` or `Entity` object into suitable target format."""
412 @staticmethod
413 @abstractmethod
414 def _to_person(person_obj: Any) -> Person | Entity | None:
415 """Convert an object representing a person into a `Person` or `Entity` object."""
417 @classmethod
418 def _parse_people(cls, people: list[Any] | None) -> list[Person | Entity]:
419 """Return a list of Persons and Entities parsed from list of format-specific people representations."""
420 # remove None values
421 return [
422 person
423 for p in people or []
424 if p is not None
425 if (person := cls._to_person(p)) is not None
426 ]
428 # ----
429 # individual magic getters and setters
431 def _get_key(self, key: str) -> str | list[str] | IgnoreKey:
432 """Get a key itself."""
433 return self.direct_mappings.get(key) or key
435 @property
436 def name(self):
437 """Return the name of the project."""
438 return self._get_property(self._get_key("name"))
440 @name.setter
441 def name(self, name: str) -> None:
442 """Set the name of the project."""
443 self._set_property(self._get_key("name"), name)
445 @property
446 def version(self) -> str | None:
447 """Return the version of the project."""
448 return self._get_property(self._get_key("version"))
450 @version.setter
451 def version(self, version: str | None) -> None:
452 """Set the version of the project."""
453 self._set_property(self._get_key("version"), version)
455 @property
456 def description(self) -> str | None:
457 """Return the description of the project."""
458 return self._get_property(self._get_key("description"))
460 @description.setter
461 def description(self, description: str) -> None:
462 """Set the description of the project."""
463 self._set_property(self._get_key("description"), description)
465 @property
466 def authors(self):
467 """Return the authors of the project."""
468 authors = self._get_property(self._get_key("authors"))
469 if authors is None or len(authors) == 0:
470 return []
472 # only return authors that can be converted to Person
473 authors_validated = [
474 author for author in authors if self._to_person(author) is not None
475 ]
476 return authors_validated
478 @authors.setter
479 def authors(self, authors: list[Person | Entity]) -> None:
480 """Set the authors of the project."""
481 authors = [self._from_person(c) for c in authors]
482 self._set_property(self._get_key("authors"), authors)
484 @property
485 def maintainers(self):
486 """Return the maintainers of the project."""
487 maintainers = self._get_property(self._get_key("maintainers"))
488 if maintainers is None:
489 return []
491 # only return maintainers that can be converted to Person
492 maintainers_validated = [
493 maintainer
494 for maintainer in maintainers
495 if self._to_person(maintainer) is not None
496 ]
497 return maintainers_validated
499 @maintainers.setter
500 def maintainers(self, maintainers: list[Person | Entity]) -> None:
501 """Set the maintainers of the project."""
502 maintainers = [self._from_person(c) for c in maintainers]
503 self._set_property(self._get_key("maintainers"), maintainers)
505 @property
506 def contributors(self):
507 """Return the contributors of the project."""
508 return self._get_property(self._get_key("contributors"))
510 @contributors.setter
511 def contributors(self, contributors: list[Person | Entity]) -> None:
512 """Set the contributors of the project."""
513 contributors = [self._from_person(c) for c in contributors]
514 self._set_property(self._get_key("contributors"), contributors)
516 @property
517 def keywords(self) -> list[str] | None:
518 """Return the keywords of the project."""
519 return self._get_property(self._get_key("keywords"))
521 @keywords.setter
522 def keywords(self, keywords: list[str]) -> None:
523 """Set the keywords of the project."""
524 self._set_property(self._get_key("keywords"), keywords)
526 @property
527 def license(self) -> Any:
528 """Return the license of the project."""
529 return self._get_property(self._get_key("license"))
531 @license.setter
532 def license(self, license: Any) -> None:
533 """Set the license of the project."""
534 self._set_property(self._get_key("license"), license)
536 @property
537 def homepage(self) -> str | None:
538 """Return the homepage url of the project."""
539 return self._get_property(self._get_key("homepage"))
541 @homepage.setter
542 def homepage(self, value: str | None) -> None:
543 """Set the homepage url of the project."""
544 self._set_property(self._get_key("homepage"), value)
546 @property
547 def repository(self) -> str | dict | None:
548 """Return the repository url of the project."""
549 return self._get_property(self._get_key("repository"))
551 @repository.setter
552 def repository(self, value: str | dict | None) -> None:
553 """Set the repository url of the project."""
554 self._set_property(self._get_key("repository"), value)
556 @property
557 def documentation(self) -> str | dict | None:
558 """Return the documentation url of the project."""
559 return self._get_property(self._get_key("documentation"))
561 @documentation.setter
562 def documentation(self, value: str | dict | None) -> None:
563 """Set the documentation url of the project."""
564 self._set_property(self._get_key("documentation"), value)