metadatarootrecord.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899
  1. from uuid import UUID
  2. from typing import (
  3. Iterable,
  4. Optional
  5. )
  6. from dataladmetadatamodel.mappableobject import MappableObject
  7. from dataladmetadatamodel.filetree import FileTree
  8. from dataladmetadatamodel.metadata import Metadata
  9. from dataladmetadatamodel.mapper.reference import Reference
  10. class MetadataRootRecord(MappableObject):
  11. def __init__(self,
  12. dataset_identifier: Optional[UUID],
  13. dataset_version: Optional[str],
  14. dataset_level_metadata: Optional[Metadata],
  15. file_tree: Optional[FileTree],
  16. reference: Optional[Reference] = None,
  17. backend_type: str = "git"):
  18. assert isinstance(dataset_identifier, (type(None), UUID))
  19. assert isinstance(dataset_version, (type(None), str))
  20. assert isinstance(dataset_level_metadata, (type(None), Metadata))
  21. assert isinstance(file_tree, (type(None), FileTree))
  22. assert isinstance(reference, (type(None), Reference))
  23. super().__init__(reference)
  24. self.dataset_identifier = dataset_identifier
  25. self.dataset_version = dataset_version
  26. self.dataset_level_metadata = dataset_level_metadata
  27. self.file_tree = file_tree
  28. self.backend_type = backend_type
  29. @staticmethod
  30. def get_empty_instance(reference: Optional[Reference] = None):
  31. return MetadataRootRecord(None, None, None, None, reference)
  32. def get_modifiable_sub_objects_impl(self) -> Iterable[MappableObject]:
  33. return [
  34. child
  35. for child in [self.dataset_level_metadata, self.file_tree]
  36. if child is not None
  37. ]
  38. def purge_impl(self):
  39. self.dataset_level_metadata.purge()
  40. if self.file_tree is not None:
  41. self.file_tree.purge()
  42. if self.dataset_level_metadata is not None:
  43. self.dataset_level_metadata = None
  44. self.file_tree = None
  45. def set_file_tree(self, file_tree: FileTree):
  46. self.touch()
  47. self.file_tree = file_tree
  48. def get_file_tree(self):
  49. self.ensure_mapped()
  50. if self.file_tree is None:
  51. return None
  52. return self.file_tree.read_in(self.backend_type)
  53. def set_dataset_level_metadata(self, dataset_level_metadata: Metadata):
  54. self.touch()
  55. self.dataset_level_metadata = dataset_level_metadata
  56. def get_dataset_level_metadata(self):
  57. self.ensure_mapped()
  58. return self.dataset_level_metadata.read_in(self.backend_type)
  59. def deepcopy_impl(self,
  60. new_mapper_family: Optional[str] = None,
  61. new_destination: Optional[str] = None,
  62. **kwargs) -> "MetadataRootRecord":
  63. copied_metadata_root_record = MetadataRootRecord(
  64. self.dataset_identifier,
  65. self.dataset_version,
  66. (
  67. self.dataset_level_metadata.deepcopy(
  68. new_mapper_family,
  69. new_destination)
  70. if self.dataset_level_metadata is not None
  71. else None
  72. ),
  73. (
  74. self.file_tree.deepcopy(
  75. new_mapper_family,
  76. new_destination)
  77. if self.file_tree is not None
  78. else None
  79. ),
  80. None,
  81. self.backend_type)
  82. copied_metadata_root_record.write_out(new_destination)
  83. copied_metadata_root_record.purge()
  84. return copied_metadata_root_record