Repository: jla-gardner/load-atoms (language: python)

Issue:
strict verification of database description files
We need a schema/some sort of formal validation of the database entry files (currently in `datasets/**/<dataset-id>.yaml`.
This is a first step towards adding more complicated fields.


Discussion / hints:

Fix patch:
diff --git a/datasets/C-GAP-17.yaml b/datasets/C-GAP-17.yaml
index 14e55b8..7274ab8 100644
--- a/datasets/C-GAP-17.yaml
+++ b/datasets/C-GAP-17.yaml
@@ -14,5 +14,5 @@ citation: |
         year = {2017},
         pages = {094203},
     }
-license: https://creativecommons.org/licenses/by-nc-sa/4.0/
-representative structures: [1, 1000, 2000]
+license: CC BY-NC-SA 4.0
+representative_structures: [1, 1000, 2000]
diff --git a/dev/DATASETS.md b/dev/DATASETS.md
index 4838d24..ce2144c 100644
--- a/dev/DATASETS.md
+++ b/dev/DATASETS.md
@@ -4,6 +4,10 @@ This project is designed to be as easy as possible to add new datasets to. To ad
 
 ## 1. Add a `<dataset>.yaml` file in the `load_atoms/datasets` directory
 
+Dataset descriptions are stored in YAML files, which are located in the `load_atoms/datasets` directory.
+These are programatically validated at runtime, so that the user can be sure that the dataset is valid.
+The blueprint against which these are validated (`DESCRIPTION_BLUEPRINT`) is located in `load_atoms/database.py`.
+
 This file should contain the following fields:
 
 -   `name`: The name of the dataset. This should be a unique, preferably short and easily readable identifier for the dataset
@@ -14,13 +18,11 @@ Optionally, you can also specify the following fields:
 
 -   `license`: The license under which the dataset is released.
 -   `citation`: A citation for the dataset, if applicable. This should be a BibTeX citation.
--   `representative structures`: a list of structures that are representative of the dataset. The first 5 of these will be used to generate the dataset thumbnail on the documentation website.
+-   `representative_structures`: a list of indexes of structures that are representative of the dataset. The first 5 of these will be used to generate the dataset thumbnail on the documentation website.
 
-Example:
+Example file located at `load_atoms/datasets/my-dataset.yaml`:
 
 ```yaml
-# load_atoms/datasets/my-dataset.yaml
-
 name: my-dataset
 description: A dataset of 1000 atoms
 filename: my-dataset.xyz
diff --git a/dev/generate_page.py b/dev/generate_page.py
index 7f3605d..92fea8f 100644
--- a/dev/generate_page.py
+++ b/dev/generate_page.py
@@ -33,24 +33,24 @@
 def page_content(dataset_id: str) -> str:
     # avoid actually downloading the dataset
     structures = dataset(dataset_id, root=PROJECT_ROOT / "datasets")
-    db_entry = DATASETS[dataset_id]
+    dataset = DATASETS[dataset_id]
 
-    representative_structures = db_entry.representative_structures or [*range(5)]
+    representative_structures = dataset.representative_structures or [*range(5)]
     representative_structures = representative_structures[: min(5, len(structures))]
 
     summary = str(structures)
 
-    title = db_entry.name + "\n" + "=" * len(db_entry.name)
+    title = dataset.name + "\n" + "=" * len(dataset.name)
     visualisations = ""
 
     return PAGE_TEMPLATE.format(
         title=title,
-        description=db_entry.description,
+        description=dataset.description,
         dataset_id=dataset_id,
         summary=pad(summary, indent=4),
         visualisations=visualisations,
-        license=db_entry.license,
-        citation=pad(db_entry.citation, indent=4),
+        license=dataset.license,
+        citation=pad(dataset.citation, indent=4),
     )
 
 
diff --git a/src/load_atoms/backend.py b/src/load_atoms/backend.py
index 23a9e80..2894065 100644
--- a/src/load_atoms/backend.py
+++ b/src/load_atoms/backend.py
@@ -10,7 +10,7 @@
 import requests
 from ase.io import read
 
-from load_atoms.database import DatabaseEntry, get_database_entry_for, print_info_for
+from load_atoms.database import DatasetDescription, get_description_of, print_info_for
 from load_atoms.dataset import Dataset
 from load_atoms.util import BASE_REMOTE_URL, DEFAULT_DOWNLOAD_DIR, progress_bar
 
@@ -25,21 +25,22 @@ def load_dataset_for(
     else:
         root = Path(root)
 
-    db_entry = get_database_entry_for(dataset_id)
+    dataset_description = get_description_of(dataset_id)
 
-    download_if_needed(db_entry, root)
-    print_info_for(db_entry)
+    download_if_needed(dataset_description, root)
+    print_info_for(dataset_description)
 
     all_structures = []
-    for filename in db_entry.filenames:
+    for filename in dataset_description.filenames:
         all_structures.extend(read(root / filename, index=":"))
-    return Dataset(all_structures, db_entry.name)
 
+    return Dataset(all_structures, dataset_description.name)
 
-def download_if_needed(db_entry: DatabaseEntry, root: Path) -> None:
+
+def download_if_needed(dataset: DatasetDescription, root: Path) -> None:
     """Download a dataset if it is not already present."""
 
-    for filename in db_entry.filenames:
+    for filename in dataset.filenames:
         local_path = root / filename
         if local_path.exists():
             continue
diff --git a/src/load_atoms/database.py b/src/load_atoms/database.py
index c781ab5..6ee4bb4 100644
--- a/src/load_atoms/database.py
+++ b/src/load_atoms/database.py
@@ -1,38 +1,79 @@
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 from pathlib import Path
-from typing import List
 
 import yaml
 
 from load_atoms.util import DATASETS_DIR
+from load_atoms.validation import (
+    AnyOf,
+    Blueprint,
+    IsBibTeX,
+    IsIn,
+    OneOf,
+    Optional,
+    Required,
+)
+
+VALID_LICENSES = ["MIT", "CC-BY-4.0", "CC BY-NC-SA 4.0"]
+
+DESCRIPTION_BLUEPRINT = Blueprint(
+    # --- required fields ---
+    # -----------------------
+    Required("name", str),
+    Required("description", str),
+    OneOf(
+        Required("filename", str),
+        Required("filenames", [str]),
+    ),
+    # --- optional fields ---
+    # -----------------------
+    Optional("citation", IsBibTeX()),
+    Optional("license", IsIn(VALID_LICENSES)),
+    Optional("representative_structures", [int]),
+    Optional("long_description", str),
+    Optional(
+        "properties",
+        AnyOf(
+            Optional("per atom", dict),
+            Optional("per structure", dict),
+        ),
+    ),
+)
 
 
 @dataclass
-class DatabaseEntry:
+class DatasetDescription:
     name: str
-    filenames: List[str]
     description: str
+    filename: str = None
+    filenames: list = None
+
     citation: str = None
     license: str = None
-    representative_structures: List[int] = None
-
-    @classmethod
-    def from_file(cls, file: Path):
-        with open(file) as f:
-            data = yaml.safe_load(f)
+    representative_structures: list = None
+    long_description: str = None
+    properties: dict = None
 
-        if "filename" in data:
-            data["filenames"] = [data["filename"]]
-            del data["filename"]
+    def __post_init__(self):
+        # validate the data
+        DESCRIPTION_BLUEPRINT.validate(asdict(self))
 
-        data = {k.replace(" ", "_"): v for k, v in data.items()}
+        # make sure we have a list of filenames
+        if self.filename is not None:
+            assert self.filenames is None
+            self.filenames = [self.filename]
 
+    @classmethod
+    def from_file(cls, path: Path) -> "DatasetDescription":
+        with open(path) as f:
+            data = yaml.safe_load(f)
+        DESCRIPTION_BLUEPRINT.validate(data)
         return cls(**data)
 
 
+_descriptor_files = list(DATASETS_DIR.glob("**/*.yaml"))
 DATASETS = {
-    entry.name: entry
-    for entry in map(DatabaseEntry.from_file, DATASETS_DIR.glob("**/*.yaml"))
+    entry.name: entry for entry in map(DatasetDescription.from_file, _descriptor_files)
 }
 
 
@@ -41,19 +82,19 @@ def is_known_dataset(dataset_id: str) -> bool:
     return dataset_id in DATASETS
 
 
-def get_database_entry_for(dataset_id: str) -> DatabaseEntry:
+def get_description_of(dataset_id: str) -> DatasetDescription:
     """Get the database entry for a dataset."""
     assert is_known_dataset(dataset_id), f"Dataset {dataset_id} is not known."
     return DATASETS[dataset_id]
 
 
-def print_info_for(db_entry: DatabaseEntry) -> None:
+def print_info_for(dataset: DatasetDescription) -> None:
     """print description and any license/citation info for a dataset"""
-    print(db_entry.description)
+    print(dataset.description)
 
-    if db_entry.license is not None:
-        print("This dataset is licensed under", db_entry.license.strip())
+    if dataset.license is not None:
+        print("This dataset is licensed under", dataset.license.strip())
 
-    if db_entry.citation is not None:
+    if dataset.citation is not None:
         print("If you use this dataset, please cite the following:")
-        print(db_entry.citation.strip())
+        print(dataset.citation.strip())
diff --git a/src/load_atoms/util.py b/src/load_atoms/util.py
index e818341..dba2866 100644
--- a/src/load_atoms/util.py
+++ b/src/load_atoms/util.py
@@ -11,6 +11,10 @@
 TODO = "░"
 
 
+def lpad(s, indent=4):
+    return "\n".join(" " * indent + line for line in s.splitlines())
+
+
 def intersection(things: Iterable[Iterable]):
     """Get the set intersection of a list of iterables."""
     return set.intersection(*map(set, things))
@@ -29,8 +33,6 @@ def _progress(count, total):
         message = f"{progress:6.1%} | {done}{todo}"
         print(message, end="\r")
 
-    print()
-
     _progress(0, N)
     for i, item in enumerate(iterable):
         yield item
diff --git a/src/load_atoms/validation.py b/src/load_atoms/validation.py
new file mode 100644
index 0000000..5d7450f
--- /dev/null
+++ b/src/load_atoms/validation.py
@@ -0,0 +1,256 @@
+from load_atoms.util import lpad
+
+
+class RuleValidationError(Exception):
+    pass
+
+
+class Rule:
+    def is_valid(self, data) -> bool:
+        """
+        check if the data is valid according to the rule
+        """
+        raise NotImplementedError()
+
+    def _describe(self, content="") -> str:
+        """
+        describe the rule
+        """
+        return f"{self.__class__.__name__}({content})"
+
+    def __repr__(self, indent=0) -> str:
+        return lpad(self._describe(), indent=indent)
+
+    def validate(self, data) -> None:
+        """
+        validate the data according to the rule
+
+        Raises:
+            ValidationError: if the data is not valid
+        """
+
+        if not self.is_valid(data):
+            raise RuleValidationError(f"Rule {self} failed on {data}")
+
+    def __call__(self, data):
+        return self.is_valid(data)
+
+
+class IsInstance(Rule):
+    """
+    check if the data is an instance of the given class
+
+    Usage:
+    >>> rule = IsInstance(str)
+    >>> rule("hello")
+    True
+    >>> rule(1)
+    False
+    """
+
+    def __init__(self, cls):
+        self.cls = cls
+
+    def is_valid(self, data):
+        return isinstance(data, self.cls)
+
+    def _describe(self):
+        # return super()._describe(self.cls.__name__)
+        return self.cls.__name__
+
+
+class MultiRule(Rule):
+    def __init__(self, *rules: Rule):
+        self.rules = rules
+
+    def apply_rules(self, data):
+        passes = [rule for rule in self.rules if rule(data)]
+        failures = [rule for rule in self.rules if not rule(data)]
+        return passes, failures
+
+    def _describe(self):
+        parts = ",\n".join(rule.__repr__(indent=4) for rule in self.rules)
+        return super()._describe(f"(\n{parts}\n)")
+
+    def is_valid(self, data) -> bool:
+        try:
+            self.validate(data)
+            return True
+        except RuleValidationError:
+            return False
+
+
+class AllOf(MultiRule):
+    def validate(self, data):
+        passes, failures = self.apply_rules(data)
+        if len(failures) == 0:
+            return
+
+        raise RuleValidationError(
+            f"Rule {self} failed on {data}.\n"
+            f" {len(failures)} rules failed:\n"
+            + "\n".join(f" - {rule}" for rule in failures)
+        )
+
+
+class Blueprint(AllOf):
+    pass
+
+
+class OneOf(MultiRule):
+    def validate(self, data):
+        passes, failures = self.apply_rules(data)
+        if len(passes) == 1:
+            return
+
+        raise RuleValidationError(
+            f"Rule {self} failed on {data}.\n"
+            f"More than one rule passed:\n" + "\n".join(f" - {rule}" for rule in passes)
+        )
+
+
+class AnyOf(MultiRule):
+    def validate(self, data):
+        passes, failures = self.apply_rules(data)
+        if len(passes) >= 1:
+            return
+
+        raise RuleValidationError(
+            f"Rule {self} failed on {data}.\n" f"No rules passed.\n"
+        )
+
+
+class ListOf(Rule):
+    """
+    check if the data is a list of valid items
+
+    Usage:
+    >>> rule = ListOf(IsInstance(int))
+    >>> rule([1, 2, 3])
+    True
+    >>> rule([1, "hello", 3])
+    False
+    >>> rule = ListOf(str)  # shorthand for ListOf(IsInstance(str))
+    >>> rule(["hello", "world"])
+    True
+    """
+
+    def __init__(self, validator: Rule):
+        self.validator = convert_from_shorthand(validator)
+
+    def is_valid(self, data):
+        if not isinstance(data, list):
+            return False
+        for item in data:
+            if not self.validator(item):
+                return False
+        return True
+
+    def _describe(self):
+        # return super()._describe(self.validator)
+        return f"[{self.validator}]"
+
+
+def convert_from_shorthand(validator):
+    if validator in (str, int, float, list, dict):
+        return IsInstance(validator)
+    if isinstance(validator, type):
+        return IsInstance(validator)
+    if isinstance(validator, list):
+        assert len(validator) == 1
+        return ListOf(validator[0])
+    return validator
+
+
+class FieldValidator(Rule):
+    def __init__(self, field, validator=None):
+        self.field = field
+        self.validator = convert_from_shorthand(validator)
+
+    def _describe(self):
+        content = f"{self.field!r}"
+        if self.validator is not None:
+            content += f", {self.validator}"
+        return super()._describe(content)
+
+
+class Required(FieldValidator):
+    """
+    check if the data has the given field
+
+    Usage:
+    >>> rule = Required("name")
+    >>> rule({"name": "hello"})
+    True
+    >>> rule({"name": 1})
+    False
+    >>> rule = Required("name", str)
+    >>> rule({"name": "hello"})
+    True
+    >>> rule({"name": 1})
+    False
+    """
+
+    def is_valid(self, data):
+        if self.field not in data:
+            return False
+        if self.validator is not None:
+            return self.validator(data[self.field])
+        return True
+
+
+class Optional(FieldValidator):
+    """
+    check if the data has the given field
+
+    Usage:
+    >>> rule = Optional("name")
+    >>> rule({"name": "hello"})
+    True
+    >>> rule({"name": 1})
+    False
+    >>> rule = Optional("name", str)
+    >>> rule({"name": "hello"})
+    True
+    >>> rule({"name": 1})
+    False
+    """
+
+    def is_valid(self, data):
+        # rule passes if the field is not present
+        if getattr(data, self.field, None) is None:
+            return True
+        if self.field not in data:
+            return True
+        # but if it is present, it must be valid
+        if self.validator is not None:
+            return self.validator(data[self.field])
+        return True
+
+
+class IsIn(Rule):
+    """
+    check if the data is in the given iterable
+
+    Usage:
+    >>> rule = IsIn({"hello", "world"})
+    >>> rule("hello")
+    True
+    >>> rule("foo")
+    False
+    """
+
+    def __init__(self, options):
+        self.options = options
+
+    def is_valid(self, data):
+        return data in self.options
+
+    def _describe(self) -> str:
+        return super()._describe(self.options)
+
+
+class IsBibTeX(Rule):
+    def is_valid(self, data):
+        data = data.strip()
+        return data.startswith("@") and data.endswith("}")


Test patch:
diff --git a/tests/test_validation.py b/tests/test_validation.py
new file mode 100644
index 0000000..8b52b8c
--- /dev/null
+++ b/tests/test_validation.py
@@ -0,0 +1,56 @@
+import pytest
+
+from load_atoms.validation import (
+    Blueprint,
+    IsBibTeX,
+    IsIn,
+    ListOf,
+    OneOf,
+    Optional,
+    Required,
+    RuleValidationError,
+)
+
+bob = {
+    "name": "bob",
+    "description": "a happy little fellow",
+    "age": 42,
+    "hobbies": ["coding", "reading", "writing"],
+}
+
+
+def test_required():
+    true_rules = [
+        Required("name"),  # exists
+        Required("description", str),  # correct type
+        Required("age", int),  # correct type
+        Required("hobbies", ListOf(str)),  # correct type
+        Required("age", lambda x: x > 0),  # custom validation
+        Required("age", IsIn([42, 43])),  # custom validation
+    ]
+
+    false_rules = [
+        Required("made_up_field"),  # doesn't exist
+        Required("name", int),  # wrong type
+    ]
+
+    # test each rule individually
+    for rule in true_rules:
+        assert rule(bob), f"rule {rule} failed"
+
+    for rule in false_rules:
+        assert not rule(bob), f"rule {rule} should have failed"
+
+    # test all rules together
+    schema = Blueprint(*true_rules)
+    schema.validate(bob)
+
+
+def test_failure():
+    rule = Required("name", int)
+    assert not rule(bob)
+
+    schema = Blueprint(rule)
+    # assert that validation fails
+    with pytest.raises(RuleValidationError):
+        schema.validate(bob)