Merge pull request #40 from lsst/tickets/DM-42446-andys-review

DM-42446: Implement minor fixes and improvements to Python syntax

python/felis/cli.py

@@ -19,12 +19,14 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+from __future__ import annotations
+
 import io
 import json
 import logging
 import sys
 from collections.abc import Iterable, Mapping, MutableMapping
-from typing import Any, Type
+from typing import Any
 
 import click
 import yaml
@@ -314,7 +316,7 @@ def merge(files: Iterable[io.TextIOBase]) -> None:
 @click.argument("files", nargs=-1, type=click.File())
 def validate(schema_name: str, require_description: bool, files: Iterable[io.TextIOBase]) -> None:
     """Validate one or more felis YAML files."""
-    schema_class: Type[Schema] = get_schema(schema_name)
+    schema_class = get_schema(schema_name)
     logger.info(f"Using schema '{schema_class.__name__}'")
 
     if require_description:

python/felis/datamodel.py

@@ -19,10 +19,12 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+from __future__ import annotations
+
 import logging
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from enum import Enum
-from typing import Any, Literal, Sequence
+from typing import Any, Literal
 
 from astropy import units as units  # type: ignore
 from astropy.io.votable import ucd  # type: ignore
@@ -91,8 +93,8 @@ def check_description(cls, values: dict[str, Any]) -> dict[str, Any]:
         if Schema.is_description_required():
             if "description" not in values or not values["description"]:
                 raise ValueError("Description is required and must be non-empty")
-            if len(values["description"].strip()) < 3:
-                raise ValueError("Description must be at least three characters long")
+            if len(values["description"].strip()) < DESCR_MIN_LENGTH:
+                raise ValueError(f"Description must be at least {DESCR_MIN_LENGTH} characters long")
         return values
 
 
@@ -418,7 +420,7 @@ def check_unique_table_names(cls, tables: list[Table]) -> list[Table]:
         return tables
 
     @model_validator(mode="after")
-    def create_id_map(self: "Schema") -> "Schema":
+    def create_id_map(self: Schema) -> Schema:
         """Create a map of IDs to objects."""
         visitor: SchemaIdVisitor = SchemaIdVisitor()
         visitor.visit_schema(self)
@@ -432,7 +434,7 @@ def create_id_map(self: "Schema") -> "Schema":
     def __getitem__(self, id: str) -> BaseObject:
         """Get an object by its ID."""
         if id not in self:
-            raise ValueError(f"Object with ID '{id}' not found in schema")
+            raise KeyError(f"Object with ID '{id}' not found in schema")
         return self.id_map[id]
 
     def __contains__(self, id: str) -> bool:
@@ -449,9 +451,9 @@ def require_description(cls, rd: bool = True) -> None:
         when validating schemas, rather than change the flag value directly.
         """
         logger.debug(f"Setting description requirement to '{rd}'")
-        Schema.ValidationConfig._require_description = rd
+        cls.ValidationConfig._require_description = rd
 
     @classmethod
     def is_description_required(cls) -> bool:
         """Return whether a description is required for all objects."""
-        return Schema.ValidationConfig._require_description
+        return cls.ValidationConfig._require_description

python/felis/utils.py

@@ -19,6 +19,8 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+from __future__ import annotations
+
 from collections.abc import Iterable, Mapping, MutableMapping
 from typing import Any
 

python/felis/validation.py

@@ -19,22 +19,26 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+from __future__ import annotations
+
 import logging
-from typing import Any, Sequence, Type
+from collections.abc import Sequence
+from typing import Any
 
 from pydantic import Field, model_validator
 
 from .datamodel import DESCR_MIN_LENGTH, Column, Schema, Table
 
 logger = logging.getLogger(__name__)
 
+__all__ = ["RspColumn", "RspSchema", "RspTable", "get_schema"]
+
 
 class RspColumn(Column):
     """Column for RSP data validation."""
 
     description: str = Field(..., min_length=DESCR_MIN_LENGTH)
-    """Redefine description to make it required and non-empty.
-    """
+    """Redefine description to make it required and non-empty."""
 
 
 class RspTable(Table):
@@ -93,7 +97,7 @@ def check_tap_table_indexes(cls: Any, sch: "RspSchema") -> "RspSchema":
         return sch
 
 
-def get_schema(schema_name: str) -> Type[Schema]:
+def get_schema(schema_name: str) -> type[Schema]:
     """Get the schema class for the given name."""
     if schema_name == "default":
         return Schema

tests/test_cli.py

@@ -137,11 +137,18 @@ def test_validate_default(self) -> None:
     def test_validate_default_with_require_description(self) -> None:
         """Test validate command with description required."""
         runner = CliRunner()
-        result = runner.invoke(cli, ["validate", "--require-description", TEST_YAML], catch_exceptions=False)
-        # This state needs to be reset for subsequent tests or some will fail.
-        # This is not an issue when running the actual command line tool, which
-        # will execute in its own system process.
-        Schema.require_description(False)
+        try:
+            # Wrap this in a try/catch in case an exception is thrown.
+            result = runner.invoke(
+                cli, ["validate", "--require-description", TEST_YAML], catch_exceptions=False
+            )
+        except Exception as e:
+            # Reraise exception.
+            raise e
+        finally:
+            # Turn the flag off so it does not effect subsequent tests.
+            Schema.require_description(False)
+
         self.assertEqual(result.exit_code, 0)
 
     def test_validate_rsp(self) -> None:

tests/test_datamodel.py

@@ -121,7 +121,9 @@ def test_validation(self) -> None:
         with self.assertRaises(ValidationError):
             Column(**units_data)
 
-        # Turn on description requirement for next two tests.
+    def test_require_description(self) -> None:
+        """Test the require_description flag for the `Column` class."""
+        # Turn on description requirement for this test.
         Schema.require_description(True)
 
         # Make sure that setting the flag for description requirement works
@@ -131,14 +133,28 @@ def test_validation(self) -> None:
         # Creating a column without a description when required should throw an
         # error.
         with self.assertRaises(ValidationError):
-            col = Column(
+            Column(
                 **{
                     "name": "testColumn",
                     "@id": "#test_col_id",
                     "datatype": "string",
                 }
             )
 
+        # Creating a column with a None description when required should throw.
+        with self.assertRaises(ValidationError):
+            Column(**{"name": "testColumn", "@id": "#test_col_id", "datatype": "string", "description": None})
+
+        # Creating a column with an empty description when required should
+        # throw.
+        with self.assertRaises(ValidationError):
+            Column(**{"name": "testColumn", "@id": "#test_col_id", "datatype": "string", "description": ""})
+
+        # Creating a column with a description that is not long enough should
+        # throw.
+        with self.assertRaises(ValidationError):
+            Column(**{"name": "testColumn", "@id": "#test_col_id", "datatype": "string", "description": "xy"})
+
         # Turn off flag or it will affect subsequent tests.
         Schema.require_description(False)
 
@@ -382,7 +398,7 @@ def test_schema_object_ids(self) -> None:
         self.assertIsInstance(sch["#test_table_id"], Table, "schema[id] should return a Table")
         self.assertIsInstance(sch["#test_schema_id"], Schema, "schema[id] should return a Schema")
 
-        with self.assertRaises(ValueError):
+        with self.assertRaises(KeyError):
             # Test that an invalid id raises an exception.
             sch["#bad_id"]
 

tests/test_validation.py

@@ -47,6 +47,10 @@ def test_rsp_validation(self) -> None:
         with self.assertRaises(ValidationError):
             RspColumn(name="testColumn", id="#test_col_id", datatype="string", description=None)
 
+        # A column description which is not long enough should throw.
+        with self.assertRaises(ValidationError):
+            RspColumn(name="testColumn", id="#test_col_id", datatype="string", description="xy")
+
         # Creating a valid RSP column should not throw an exception.
         col = RspColumn(
             **{