Coverage for metacell/storage/metadata.py : 71%

""" 

Manage meta-data in YAML files. 

""" 

from tgutils.cache import Cache 

from tgutils.load_yaml import load_dictionary 

from types import SimpleNamespace 

from typing import Any 

from typing import Dict 

from typing import List 

from typing import Type 

from typing import TypeVar 

from uuid import UUID 

import os 

import yaml 

YM = TypeVar('YM', bound='YamlMetadata') 

class YamlMetadata(SimpleNamespace): 

""" 

Meta-data stored in a YAML file. 

""" 

#: The keys that must appear in the YAML file. 

required_keys: Dict[str, Any] = dict(uuid=str) 

#: The keys that should be removed from the YAML file. 

removed_keys: List[str] = ['path', 'yaml_path'] 

_cache: Cache[str, 'YamlMetadata'] = Cache() 

def __init__(self, **kwargs: Any) -> None: 

""" 

Create the metadata object. 

""" 

super().__init__(**kwargs) 

#: The globally unique identifier of the data. 

self.uuid: UUID 

#: The path of the directory containing the data. 

self.path: str 

#: The path of the YAML meta-data file. 

self.yaml_path: str 

assert os.path.dirname(self.yaml_path) == self.path 

def as_dict(self, **kwargs: Any) -> Dict[str, Any]: 

""" 

Return the metadata as a clean dictionary for dumping to YAML. 

""" 

dictionary = vars(self).copy() 

dictionary.update(kwargs) 

for key in self.removed_keys: 

if key in dictionary: 

del dictionary[key] 

for key, value in dictionary.items(): 

if key.endswith('uuid'): 

dictionary[key] = str(value) 

return dictionary 

def dump(self, yaml_path: str, **kwargs: Any) -> None: 

""" 

Dump batch metadata into a file. 

""" 

with open(yaml_path, 'w') as file: 

file.write(yaml.dump(self.as_dict(**kwargs), default_flow_style=False)) 

@classmethod 

def load(cls: Type[YM], yaml_path: str) -> YM: 

""" 

Load the batch meta-data from a file. 

""" 

return cls._cache.lookup(yaml_path, # type: ignore 

lambda: YamlMetadata._load(cls, yaml_path)) 

@staticmethod 

def _load(cls: Type[YM], yaml_path: str) -> YM: # pylint: disable=bad-staticmethod-argument 

dictionary = load_dictionary(yaml_path) 

cls = cls.detect_cls(cls, dictionary) 

dictionary = load_dictionary(yaml_path, dictionary, required_keys=cls.required_keys) 

for key, value in dictionary.items(): 

if key.endswith('uuid'): 

dictionary[key] = UUID(value) 

dictionary['yaml_path'] = yaml_path 

dictionary['path'] = os.path.dirname(yaml_path) 

return cls(**dictionary) 

@staticmethod 

def detect_cls(cls: Type[YM], # pylint: disable=bad-staticmethod-argument 

_dictionary: Dict[str, Any]) -> Type[YM]: 

""" 

Decide on the concrete metadata class to load from the dictionary. 

""" 

return cls