diff --git a/CHANGELOG.md b/CHANGELOG.md index 9873b7692..d4413ed0e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # HDMF Changelog +## HDMF 6.2.0 (Upcoming) + +### Enhancements +- Added the `hdmf.typing` package, the first phase of replacing `@docval` with standard Python type hints ([#1129](https://github.com/hdmf-dev/hdmf/issues/1129)). Runtime validation is built entirely on the type-hint system: type checks run through beartype and array dtype/shape checks through numpydantic `NDArray[...]` hints. The package provides type aliases that are real beartype validators, enforceable by `@beartype` or `beartype.door.is_bearable` anywhere (`Int`, `UInt`, `Float`, `Bool` accepting numpy scalar types — a bare `int` hint is strict; `ArrayData`, `ScalarData`, `AnyData` backed by a live type registry; `TypeName[...]` for MRO-name forward references; `Shaped[...]` for array shape specs), the `@validated` decorator (docval-style error messages, `TermSetWrapper` unwrapping, `HDMF_TYPE_CHECKING=off` kill switch), a migration tool (`python -m hdmf.typing.migrate`), and parity-testing helpers (`hdmf.typing.testing`). @bendichter [#TBD] +- `hdmf.utils.get_docval` now also works on functions with type-hinted signatures: docval-compatible argument specs are synthesized from the signature, type hints, and Google-style docstring. This keeps the downstream splice pattern `@docval(*get_docval(Parent.__init__, ...))` working as HDMF (and downstream libraries) migrate off `@docval` module by module. Existing `@docval`-decorated functions are unaffected. @bendichter [#TBD] +- New required dependencies: `beartype`, `numpydantic`, and `docstring_parser`. @bendichter [#TBD] +- Migrated `hdmf.data_utils`, `hdmf.query`, `hdmf.validate.validator`, `hdmf.common.sparse`, `hdmf.common.resources`, and `hdmf.common.hierarchicaltable` off `@docval` to type-hinted signatures with `@validated` (includes `GenericDataChunkIterator`, `DataChunkIterator`, `DataChunk`, `DataIO`, `CSRMatrix`, `HERD`, and the validator classes). Downstream code is unaffected: `get_docval` on these functions returns equivalent specs synthesized from the type hints, so patterns like `@docval(*get_docval(DataIO.__init__), ...)` (used by `H5DataIO` and extensions) keep working, positional/keyword calling conventions are unchanged, and error messages keep the docval format. @bendichter [#TBD] + ## HDMF 6.1.0 (June 25, 2026) ### Enhancements diff --git a/docs/source/api_docs.rst b/docs/source/api_docs.rst index 119c86401..831d4304e 100644 --- a/docs/source/api_docs.rst +++ b/docs/source/api_docs.rst @@ -15,6 +15,7 @@ API Documentation I/O layer Data I/O utilities Development utilities + Type hints and runtime validation Validation utilities Testing utilities Full list of HDMF package contents diff --git a/docs/source/conf.py b/docs/source/conf.py index 3b2b18fa0..f35ef939a 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -92,6 +92,16 @@ ('py:class', 'unittest.case.TestCase'), ('py:class', 'pandas.ExtensionArray'), ] +# numpy scalar types appear in expanded hdmf.typing alias signatures but have no +# py:class intersphinx targets; same for pandas internal class paths +nitpick_ignore_regex = [ + ('py:class', r'numpy\.(u?int|float|bool|longdouble)\w*'), + ('py:class', r'pandas\.core\..*'), + # beartype validators inside hdmf.typing alias signatures render as + # Is[is_array_data], Is[has_shape_...], etc., which are not documented targets + ('py:class', r'beartype\..*'), + ('py:class', r'(is|has_shape)_\w*'), +] suppress_warnings = ["config.cache"] diff --git a/pyproject.toml b/pyproject.toml index ed915fb65..2d11114fd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,9 +30,12 @@ classifiers = [ "Topic :: Scientific/Engineering :: Medical Science Apps.", ] dependencies = [ + "beartype>=0.19.0", + "docstring_parser>=0.15", "h5py>=3.6.0", "jsonschema>=3.2.0", "numpy>=1.22.0", + "numpydantic>=1.6.0", "pandas>=1.4.0", "ruamel.yaml>=0.16", ] diff --git a/src/hdmf/common/hierarchicaltable.py b/src/hdmf/common/hierarchicaltable.py index 8322d2d73..6bb79ea71 100644 --- a/src/hdmf/common/hierarchicaltable.py +++ b/src/hdmf/common/hierarchicaltable.py @@ -6,17 +6,12 @@ import numpy as np from hdmf.common.table import DynamicTable, DynamicTableRegion, VectorIndex from hdmf.common.alignedtable import AlignedDynamicTable -from hdmf.utils import docval, getargs +from hdmf.typing import Bool, Int, validated -@docval({'name': 'dynamic_table', 'type': DynamicTable, - 'doc': 'DynamicTable object to be converted to a hierarchical pandas.Dataframe'}, - returns="Hierarchical pandas.DataFrame with usually a pandas.MultiIndex on both the index and columns.", - rtype='pandas.DataFrame', - is_method=False) -def to_hierarchical_dataframe(dynamic_table): - """ - Create a hierarchical pandas.DataFrame that represents all data from a collection of linked DynamicTables. +@validated +def to_hierarchical_dataframe(dynamic_table: DynamicTable) -> pd.DataFrame: + """Create a hierarchical pandas.DataFrame that represents all data from a collection of linked DynamicTables. **LIMITATIONS:** Currently this function only supports DynamicTables with a single DynamicTableRegion column. If a table has more than one DynamicTableRegion column then the function will expand only the @@ -29,6 +24,12 @@ def to_hierarchical_dataframe(dynamic_table): * pandas.DataFrame.reset_index to turn the data from the pandas.MultiIndex into columns * :py:meth:`~hdmf.common.hierarchicaltable.drop_id_columns` to remove all 'id' columns * :py:meth:`~hdmf.common.hierarchicaltable.flatten_column_index` to flatten the column index + + Args: + dynamic_table: DynamicTable object to be converted to a hierarchical pandas.Dataframe + + Returns: + Hierarchical pandas.DataFrame with usually a pandas.MultiIndex on both the index and columns. """ # TODO: Need to deal with the case where we have more than one DynamicTableRegion column in a given table # Get the references column @@ -194,16 +195,9 @@ def __flatten_column_name(col): return col -@docval({'name': 'dataframe', 'type': pd.DataFrame, - 'doc': 'Pandas dataframe to update (usually generated by the to_hierarchical_dataframe function)'}, - {'name': 'inplace', 'type': 'bool', 'doc': 'Update the dataframe inplace or return a modified copy', - 'default': False}, - returns="pandas.DataFrame with the id columns removed", - rtype='pandas.DataFrame', - is_method=False) -def drop_id_columns(**kwargs): - """ - Drop all columns named 'id' from the table. +@validated +def drop_id_columns(dataframe: pd.DataFrame, inplace: Bool = False) -> pd.DataFrame: + """Drop all columns named 'id' from the table. In case a column name is a tuple the function will drop any column for which the inner-most name is 'id'. The 'id' columns of DynamicTable is in many cases @@ -211,8 +205,14 @@ def drop_id_columns(**kwargs): all those columns. :raises TypeError: In case that dataframe parameter is not a pandas.Dataframe. + + Args: + dataframe: Pandas dataframe to update (usually generated by the to_hierarchical_dataframe function) + inplace: Update the dataframe inplace or return a modified copy + + Returns: + pandas.DataFrame with the id columns removed """ - dataframe, inplace = getargs('dataframe', 'inplace', kwargs) col_name = 'id' drop_labels = [] for col in dataframe.columns: @@ -222,22 +222,11 @@ def drop_id_columns(**kwargs): return dataframe if inplace else re -@docval({'name': 'dataframe', 'type': pd.DataFrame, - 'doc': 'Pandas dataframe to update (usually generated by the to_hierarchical_dataframe function)'}, - {'name': 'max_levels', 'type': (int, np.integer), - 'doc': 'Maximum number of levels to use in the resulting column Index. NOTE: When ' - 'limiting the number of levels the function simply removes levels from the ' - 'beginning. As such, removing levels may result in columns with duplicate names.' - 'Value must be >0.', - 'default': None}, - {'name': 'inplace', 'type': 'bool', 'doc': 'Update the dataframe inplace or return a modified copy', - 'default': False}, - returns="pandas.DataFrame with a regular pandas.Index columns rather and a pandas.MultiIndex", - rtype='pandas.DataFrame', - is_method=False) -def flatten_column_index(**kwargs): - """ - Flatten the column index of a pandas DataFrame. +@validated +def flatten_column_index(dataframe: pd.DataFrame, + max_levels: Int | np.integer | None = None, + inplace: Bool = False) -> pd.DataFrame: + """Flatten the column index of a pandas DataFrame. The functions changes the dataframe.columns from a pandas.MultiIndex to a normal Index, with each column usually being identified by a tuple of strings. This function is @@ -246,8 +235,17 @@ def flatten_column_index(**kwargs): :raises ValueError: In case the num_levels is not >0 :raises TypeError: In case that dataframe parameter is not a pandas.Dataframe. + + Args: + dataframe: Pandas dataframe to update (usually generated by the to_hierarchical_dataframe function) + max_levels: Maximum number of levels to use in the resulting column Index. NOTE: When limiting + the number of levels the function simply removes levels from the beginning. As such, removing + levels may result in columns with duplicate names. Value must be >0. + inplace: Update the dataframe inplace or return a modified copy + + Returns: + pandas.DataFrame with a regular pandas.Index columns rather and a pandas.MultiIndex """ - dataframe, max_levels, inplace = getargs('dataframe', 'max_levels', 'inplace', kwargs) if max_levels is not None and max_levels <= 0: raise ValueError('max_levels must be greater than 0') # Compute the new column names diff --git a/src/hdmf/common/resources.py b/src/hdmf/common/resources.py index f4e37c8d6..8d7c8f9f9 100644 --- a/src/hdmf/common/resources.py +++ b/src/hdmf/common/resources.py @@ -5,7 +5,7 @@ from ..container import Table, Row, Container, Data, AbstractContainer, HERDManager from ..term_set import TermSet from ..data_utils import DataIO -from ..utils import docval, popargs, AllowPositional +from ..utils import AllowPositional from ..build import TypeMap from ..term_set import TermSetWrapper from glob import glob @@ -13,6 +13,7 @@ import zipfile from collections import namedtuple from warnings import warn +from ..typing import Bool, Int, validated class KeyTable(Table): @@ -173,31 +174,35 @@ class HERD(Container): {'name': 'entities', 'child': True}, ) - @docval({'name': 'keys', 'type': KeyTable, 'default': None, - 'doc': 'The table storing user keys for referencing resources.'}, - {'name': 'files', 'type': FileTable, 'default': None, - 'doc': 'The table for storing file ids used in external resources.'}, - {'name': 'entities', 'type': EntityTable, 'default': None, - 'doc': 'The table storing entity information.'}, - {'name': 'objects', 'type': ObjectTable, 'default': None, - 'doc': 'The table storing object information.'}, - {'name': 'object_keys', 'type': ObjectKeyTable, 'default': None, - 'doc': 'The table storing object-key relationships.'}, - {'name': 'entity_keys', 'type': EntityKeyTable, 'default': None, - 'doc': 'The table storing entity-key relationships.'}, - {'name': 'type_map', 'type': TypeMap, 'default': None, - 'doc': 'The type map. If None is provided, the HDMF-common type map will be used.'}, - allow_positional=AllowPositional.WARNING) - def __init__(self, **kwargs): + @validated(allow_positional=AllowPositional.WARNING) + def __init__(self, + keys: KeyTable | None = None, + files: FileTable | None = None, + entities: EntityTable | None = None, + objects: ObjectTable | None = None, + object_keys: ObjectKeyTable | None = None, + entity_keys: EntityKeyTable | None = None, + type_map: TypeMap | None = None): + """Initialize the HERD container. + + Args: + keys: The table storing user keys for referencing resources. + files: The table for storing file ids used in external resources. + entities: The table storing entity information. + objects: The table storing object information. + object_keys: The table storing object-key relationships. + entity_keys: The table storing entity-key relationships. + type_map: The type map. If None is provided, the HDMF-common type map will be used. + """ name = 'external_resources' super().__init__(name) - self.keys = kwargs['keys'] or KeyTable() - self.files = kwargs['files'] or FileTable() - self.entities = kwargs['entities'] or EntityTable() - self.objects = kwargs['objects'] or ObjectTable() - self.object_keys = kwargs['object_keys'] or ObjectKeyTable() - self.entity_keys = kwargs['entity_keys'] or EntityKeyTable() - self.type_map = kwargs['type_map'] or get_type_map() + self.keys = keys or KeyTable() + self.files = files or FileTable() + self.entities = entities or EntityTable() + self.objects = objects or ObjectTable() + self.object_keys = object_keys or ObjectKeyTable() + self.entity_keys = entity_keys or EntityKeyTable() + self.type_map = type_map or get_type_map() @staticmethod def assert_external_resources_equal(left, right, check_dtype=True): @@ -228,10 +233,9 @@ def assert_external_resources_equal(left, right, check_dtype=True): raise AssertionError(msg) return True - @docval({'name': 'key_name', 'type': str, 'doc': 'The name of the key to be added.'}) - def _add_key(self, **kwargs): - """ - Add a key to be used for making references to external resources. + @validated + def _add_key(self, key_name: str): + """Add a key to be used for making references to external resources. It is possible to use the same *key_name* to refer to different resources so long as the *key_name* is not used within the same object, relative_path, and field. To do so, this method must be called for the @@ -239,52 +243,53 @@ def _add_key(self, **kwargs): The returned Key objects must be managed by the caller so as to be appropriately passed to subsequent calls to methods for storing information about the different resources. + + Args: + key_name: The name of the key to be added. """ - key = kwargs['key_name'] + key = key_name return Key(key, table=self.keys) - @docval({'name': 'file_object_id', 'type': str, 'doc': 'The id of the file'}) - def _add_file(self, **kwargs): - """ - Add a file to be used for making references to external resources. + @validated + def _add_file(self, file_object_id: str): + """Add a file to be used for making references to external resources. This is optional when working in HDMF. + + Args: + file_object_id: The id of the file """ - file_object_id = kwargs['file_object_id'] return File(file_object_id, table=self.files) - @docval({'name': 'entity_id', 'type': str, 'doc': 'The unique entity id.'}, - {'name': 'entity_uri', 'type': str, 'doc': 'The URI for the entity.'}) - def _add_entity(self, **kwargs): - """ - Add an entity that will be referenced to using keys specified in HERD.entity_keys. + @validated + def _add_entity(self, entity_id: str, entity_uri: str): + """Add an entity that will be referenced to using keys specified in HERD.entity_keys. + + Args: + entity_id: The unique entity id. + entity_uri: The URI for the entity. """ - entity_id = kwargs['entity_id'] - entity_uri = kwargs['entity_uri'] entity = Entity( entity_id, entity_uri, table=self.entities) return entity - @docval({'name': 'container', 'type': (str, AbstractContainer), - 'doc': 'The Container/Data object to add or the object id of the Container/Data object to add.'}, - {'name': 'files_idx', 'type': (int, np.integer), - 'doc': 'The file_object_id row idx.'}, - {'name': 'object_type', 'type': str, 'default': None, - 'doc': ('The type of the object. This is also the parent in relative_path. If omitted, ' - 'the name of the container class is used.')}, - {'name': 'relative_path', 'type': str, - 'doc': ('The relative_path of the attribute of the object that uses ', - 'an external resource reference key. Use an empty string if not applicable.')}, - {'name': 'field', 'type': str, 'default': '', - 'doc': ('The field of the compound data type using an external resource.')}) - def _add_object(self, **kwargs): + @validated + def _add_object(self, + container: str | AbstractContainer, + files_idx: Int | np.integer, + relative_path: str, + object_type: str | None = None, + field: str = ''): + """Add an object that references an external resource. + + Args: + container: The Container/Data object to add or the object id of the Container/Data object to add. + files_idx: The file_object_id row idx. + object_type: The type of the object. This is also the parent in relative_path. If omitted, the name of the + container class is used. + relative_path: The relative_path of the attribute of the object that uses an external resource reference + key. Use an empty string if not applicable. + field: The field of the compound data type using an external resource. """ - Add an object that references an external resource. - """ - files_idx, container, object_type, relative_path, field = popargs('files_idx', - 'container', - 'object_type', - 'relative_path', - 'field', kwargs) if object_type is None: object_type = container.__class__.__name__ @@ -294,23 +299,25 @@ def _add_object(self, **kwargs): obj = Object(files_idx, container, object_type, relative_path, field, table=self.objects) return obj - @docval({'name': 'obj', 'type': (int, np.integer, Object), 'doc': 'The Object that uses the Key.'}, - {'name': 'key', 'type': (int, np.integer, Key), 'doc': 'The Key that the Object uses.'}) - def _add_object_key(self, **kwargs): - """ - Specify that an object (i.e. container and relative_path) uses a key to reference + @validated + def _add_object_key(self, obj: Int | np.integer | Object, key: Int | np.integer | Key): + """Specify that an object (i.e. container and relative_path) uses a key to reference an external resource. + + Args: + obj: The Object that uses the Key. + key: The Key that the Object uses. """ - obj, key = popargs('obj', 'key', kwargs) return ObjectKey(obj, key, table=self.object_keys) - @docval({'name': 'entity', 'type': (int, np.integer, Entity), 'doc': 'The Entity associated with the Key.'}, - {'name': 'key', 'type': (int, np.integer, Key), 'doc': 'The Key that the connected to the Entity.'}) - def _add_entity_key(self, **kwargs): - """ - Add entity-key relationship to the EntityKeyTable. + @validated + def _add_entity_key(self, entity: Int | np.integer | Entity, key: Int | np.integer | Key): + """Add entity-key relationship to the EntityKeyTable. + + Args: + entity: The Entity associated with the Key. + key: The Key that the connected to the Entity. """ - entity, key = popargs('entity', 'key', kwargs) return EntityKey(entity, key, table=self.entity_keys) def _find_object(self, file, container, relative_path, field): @@ -365,14 +372,14 @@ def _find_or_add_object(self, file, container, relative_path, field): return self._add_object(files_idx=files_idx, container=container, relative_path=relative_path, field=field) - @docval({'name': 'container', 'type': (str, AbstractContainer), - 'doc': ('The Container/Data object that uses the key or ' - 'the object id for the Container/Data object that uses the key.')}) - def _get_file_from_container(self, **kwargs): - """ - Method to retrieve a file associated with the container in the case a file is not provided. + @validated + def _get_file_from_container(self, container: str | AbstractContainer): + """Method to retrieve a file associated with the container in the case a file is not provided. + + Args: + container: The Container/Data object that uses the key or the object id for the Container/Data object that + uses the key. """ - container = kwargs['container'] if isinstance(container, HERDManager): return container @@ -394,16 +401,16 @@ def _get_file_from_container(self, **kwargs): "to the file before adding an external reference." % getattr(container, 'name', container)) raise ValueError(msg) - @docval({'name': 'objects', 'type': list, - 'doc': 'List of objects to check for TermSetWrapper within the fields.'}) - def __check_termset_wrapper(self, **kwargs): - """ - Takes a list of objects and checks the fields for TermSetWrapper. + @validated + def __check_termset_wrapper(self, objects: list): + """Takes a list of objects and checks the fields for TermSetWrapper. wrapped_obj = namedtuple('wrapped_obj', ['object', 'attribute', 'wrapper']) :return: [wrapped_obj(object1, attribute_name1, wrapper1), ...] + + Args: + objects: List of objects to check for TermSetWrapper within the fields. """ - objects = kwargs['objects'] ret = [] # list to be returned with the objects, attributes and corresponding termsets @@ -419,15 +426,15 @@ def __check_termset_wrapper(self, **kwargs): return ret - @docval({'name': 'root_container', 'type': HERDManager, - 'doc': 'The root container or file containing objects with a TermSet.'}) - def add_ref_container(self, **kwargs): - """ - Method to search through the root_container for all instances of TermSet. + @validated + def add_ref_container(self, root_container: HERDManager): + """Method to search through the root_container for all instances of TermSet. Currently, only datasets are supported. By using a TermSet, the data comes validated and can use the permissible values within the set to populate HERD. + + Args: + root_container: The root container or file containing objects with a TermSet. """ - root_container = kwargs['root_container'] all_objects = root_container.all_children() # list of child objects and the container itself @@ -449,30 +456,26 @@ def add_ref_container(self, **kwargs): entity_id=entity_id, entity_uri=entity_uri) - @docval({'name': 'container', 'type': AbstractContainer, 'default': None, - 'doc': 'The Container/Data object that uses the key.'}, - {'name': 'attribute', 'type': str, - 'doc': 'The attribute of the container for the external reference.', 'default': None}, - {'name': 'field', 'type': str, 'default': '', - 'doc': ('The field of the compound data type using an external resource.')}, - {'name': 'key', 'type': (str, Key), 'default': None, - 'doc': 'The name of the key or the Key object from the KeyTable for the key to add a resource for.'}, - {'name': 'termset', 'type': TermSet, - 'doc': 'The TermSet to be used if the container/attribute does not have one.'} - ) - def add_ref_termset(self, **kwargs): - """ - This method allows users to take advantage of using the TermSet class to provide the entity information + @validated + def add_ref_termset(self, + termset: TermSet, + container: AbstractContainer | None = None, + attribute: str | None = None, + field: str = '', + key: str | Key | None = None): + """This method allows users to take advantage of using the TermSet class to provide the entity information for add_ref, while also validating the data. This method supports adding a single key or an entire dataset to the HERD tables. For both cases, the term, i.e., key, will be validated against the permissible values in the TermSet. If valid, it will proceed to call add_ref. Otherwise, the method will return a dict of missing terms (terms not found in the TermSet). + + Args: + container: The Container/Data object that uses the key. + attribute: The attribute of the container for the external reference. + field: The field of the compound data type using an external resource. + key: The name of the key or the Key object from the KeyTable for the key to add a resource for. + termset: The TermSet to be used if the container/attribute does not have one. """ - container = kwargs['container'] - attribute = kwargs['attribute'] - key = kwargs['key'] - field = kwargs['field'] - termset = kwargs['termset'] # if key is provided then add_ref proceeds as normal if key is not None: @@ -547,38 +550,35 @@ def _resolve_object_target(self, container, attribute): return container, relative_path - @docval({'name': 'container', 'type': AbstractContainer, 'default': None, - 'doc': 'The Container/Data object that uses the key.'}, - {'name': 'attribute', 'type': str, - 'doc': 'The attribute of the container for the external reference.', 'default': None}, - {'name': 'field', 'type': str, 'default': '', - 'doc': ('The field of the compound data type using an external resource.')}, - {'name': 'key', 'type': (str, Key), 'default': None, - 'doc': ('The name of the key or the Key object from the KeyTable for the key to add a resource for. ' - 'If not provided and ``attribute`` names a scalar string attribute, the value of that ' - 'attribute is used as the key.')}, - {'name': 'entity_id', 'type': str, 'doc': 'The identifier for the entity at the resource.'}, - {'name': 'entity_uri', 'type': str, 'doc': 'The URI for the identifier at the resource.', 'default': None}, - ) - def add_ref(self, **kwargs): # noqa: C901 - """ - Add information about an external reference used in this file. + @validated + def add_ref(self, # noqa: C901 + entity_id: str, + container: AbstractContainer | None = None, + attribute: str | None = None, + field: str = '', + key: str | Key | None = None, + entity_uri: str | None = None): + """Add information about an external reference used in this file. It is possible to use the same name of the key to refer to different resources so long as the name of the key is not used within the same object, relative_path, and field combination. This method does not support such functionality by default. + + Args: + container: The Container/Data object that uses the key. + attribute: The attribute of the container for the external reference. + field: The field of the compound data type using an external resource. + key: The name of the key or the Key object from the KeyTable for the key to add a resource for. If not + provided and ``attribute`` names a scalar string attribute, the value of that attribute is used as the + key. + entity_id: The identifier for the entity at the resource. + entity_uri: The URI for the identifier at the resource. """ ############################################################### - container = kwargs['container'] - attribute = kwargs['attribute'] if isinstance(container, Data): # Used when using the TermSetWrapper if attribute == 'data': attribute = None - key = kwargs['key'] - field = kwargs['field'] - entity_id = kwargs['entity_id'] - entity_uri = kwargs['entity_uri'] ########################################## # Default the key from a scalar attribute @@ -718,30 +718,30 @@ def add_ref(self, **kwargs): # noqa: C901 if add_entity_key: self._add_entity_key(entity, key) - @docval({'name': 'key_name', 'type': str, 'doc': 'The name of the Key to get.'}, - {'name': 'file', 'type': HERDManager, 'doc': 'The file associated with the container.', - 'default': None}, - {'name': 'container', 'type': AbstractContainer, 'default': None, - 'doc': 'The Container/Data object that uses the key.'}, - {'name': 'relative_path', 'type': str, - 'doc': ('The relative_path of the attribute of the object that uses ', - 'an external resource reference key. Use an empty string if not applicable.'), - 'default': ''}, - {'name': 'field', 'type': str, 'default': '', - 'doc': ('The field of the compound data type using an external resource.')}) - def get_key(self, **kwargs): - """ - Return a Key. + @validated + def get_key(self, + key_name: str, + file: HERDManager | None = None, + container: AbstractContainer | None = None, + relative_path: str = '', + field: str = ''): + """Return a Key. If container, relative_path, and field are provided, the Key that corresponds to the given name of the key for the given container, relative_path, and field is returned. If there are multiple matches, a list of all matching keys will be returned. + + Args: + key_name: The name of the Key to get. + file: The file associated with the container. + container: The Container/Data object that uses the key. + relative_path: The relative_path of the attribute of the object that uses an external resource reference + key. Use an empty string if not applicable. + field: The field of the compound data type using an external resource. """ - key_name, container, relative_path, field = popargs('key_name', 'container', 'relative_path', 'field', kwargs) key_idx_matches = self.keys.which(key=key_name) - file = kwargs['file'] if container is not None: if file is None: @@ -766,34 +766,35 @@ def get_key(self, **kwargs): else: return self.keys.row[key_idx_matches[0]] - @docval({'name': 'entity_id', 'type': str, 'doc': 'The ID for the identifier at the resource.'}) - def get_entity(self, **kwargs): - entity_id = kwargs['entity_id'] + @validated + def get_entity(self, entity_id: str): + """get_entity + + Args: + entity_id: The ID for the identifier at the resource. + """ entity = self.entities.which(entity_id=entity_id) if len(entity)>0: return self.entities.row[entity[0]] else: return None - @docval({'name': 'object_type', 'type': str, - 'doc': 'The type of the object. This is also the parent in relative_path.'}, - {'name': 'relative_path', 'type': str, - 'doc': ('The relative_path of the attribute of the object that uses ', - 'an external resource reference key. Use an empty string if not applicable.'), - 'default': ''}, - {'name': 'field', 'type': str, 'default': '', - 'doc': ('The field of the compound data type using an external resource.')}, - {'name': 'all_instances', 'type': bool, 'default': False, - 'doc': ('The bool to return a dataframe with all instances of the object_type.', - 'If True, relative_path and field inputs will be ignored.')}) - def get_object_type(self, **kwargs): - """ - Get all entities/resources associated with an object_type. + @validated + def get_object_type(self, + object_type: str, + relative_path: str = '', + field: str = '', + all_instances: Bool = False): + """Get all entities/resources associated with an object_type. + + Args: + object_type: The type of the object. This is also the parent in relative_path. + relative_path: The relative_path of the attribute of the object that uses an external resource reference + key. Use an empty string if not applicable. + field: The field of the compound data type using an external resource. + all_instances: The bool to return a dataframe with all instances of the object_type. If True, + relative_path and field inputs will be ignored. """ - object_type = kwargs['object_type'] - relative_path = kwargs['relative_path'] - field = kwargs['field'] - all_instances = kwargs['all_instances'] df = self.to_dataframe() @@ -805,27 +806,23 @@ def get_object_type(self, **kwargs): & (df['field'] == field)] return df - @docval({'name': 'file', 'type': HERDManager, 'doc': 'The file.', - 'default': None}, - {'name': 'container', 'type': AbstractContainer, - 'doc': 'The Container/data object that is linked to resources/entities.'}, - {'name': 'attribute', 'type': str, - 'doc': 'The attribute of the container for the external reference.', 'default': None}, - {'name': 'relative_path', 'type': str, - 'doc': ('The relative_path of the attribute of the object that uses ', - 'an external resource reference key. Use an empty string if not applicable.'), - 'default': ''}, - {'name': 'field', 'type': str, 'default': '', - 'doc': ('The field of the compound data type using an external resource.')}) - def get_object_entities(self, **kwargs): + @validated + def get_object_entities(self, + container: AbstractContainer, + file: HERDManager | None = None, + attribute: str | None = None, + relative_path: str = '', + field: str = ''): + """Get all entities/resources associated with an object. + + Args: + file: The file. + container: The Container/data object that is linked to resources/entities. + attribute: The attribute of the container for the external reference. + relative_path: The relative_path of the attribute of the object that uses an external resource reference + key. Use an empty string if not applicable. + field: The field of the compound data type using an external resource. """ - Get all entities/resources associated with an object. - """ - file = kwargs['file'] - container = kwargs['container'] - attribute = kwargs['attribute'] - relative_path = kwargs['relative_path'] - field = kwargs['field'] if file is None: file = self._get_file_from_container(container=container) @@ -855,20 +852,21 @@ def get_object_entities(self, **kwargs): df = pd.DataFrame(entities, columns=['entity_id', 'entity_uri']) return df - @docval({'name': 'use_categories', 'type': bool, 'default': False, - 'doc': 'Use a multi-index on the columns to indicate which category each column belongs to.'}, - rtype='pandas.DataFrame', returns='A DataFrame with all data merged into a flat, denormalized table.') - def to_dataframe(self, **kwargs): - """ - Convert the data from the keys, resources, entities, objects, and object_keys tables + @validated + def to_dataframe(self, use_categories: Bool = False) -> pd.DataFrame: + """Convert the data from the keys, resources, entities, objects, and object_keys tables to a single joint dataframe. I.e., here data is being denormalized, e.g., keys that are used across multiple entities or objects will duplicated across the corresponding rows. Returns: :py:class:`~pandas.DataFrame` with all data merged into a single, flat, denormalized table. + Args: + use_categories: Use a multi-index on the columns to indicate which category each column belongs to. + + Returns: + A DataFrame with all data merged into a flat, denormalized table. """ - use_categories = popargs('use_categories', kwargs) # Step 1: Combine the entities, keys, and entity_keys table ent_key_df = self.entity_keys.to_dataframe() entities_mapped_df = self.entities.to_dataframe().iloc[ent_key_df['entities_idx']].reset_index(drop=True) @@ -973,12 +971,14 @@ def _repr_html_(self): html_repr += "" return html_repr - @docval({'name': 'path', 'type': str, 'doc': 'The path to the zip file.'}) - def to_zip(self, **kwargs): - """ - Write the tables in HERD to zipped tsv files. + @validated + def to_zip(self, path: str): + """Write the tables in HERD to zipped tsv files. + + Args: + path: The path to the zip file. """ - zip_file = kwargs['path'] + zip_file = path directory = os.path.dirname(zip_file) files = [os.path.join(directory, child.name)+'.tsv' for child in self.children] @@ -995,23 +995,26 @@ def to_zip(self, **kwargs): os.remove(file) @classmethod - @docval({'name': 'path', 'type': str, 'doc': 'The path to the zip file.'}) - def get_zip_directory(cls, path): - """ - Return the directory of the file given. + @validated + def get_zip_directory(cls, path: str): + """Return the directory of the file given. + + Args: + path: The path to the zip file. """ directory = os.path.dirname(os.path.realpath(path)) return directory @classmethod - @docval({'name': 'path', 'type': str, 'doc': 'The path to the zip file.'}, - {'name': 'type_map', 'type': TypeMap, 'default': None, - 'doc': 'The TypeMap to use for the returned HERD. If None, the default TypeMap is used.'}) - def from_zip(cls, **kwargs): - """ - Method to read in zipped tsv files to populate HERD. + @validated + def from_zip(cls, path: str, type_map: TypeMap | None = None): + """Method to read in zipped tsv files to populate HERD. + + Args: + path: The path to the zip file. + type_map: The TypeMap to use for the returned HERD. If None, the default TypeMap is used. """ - zip_file, type_map = popargs('path', 'type_map', kwargs) + zip_file = path directory = cls.get_zip_directory(zip_file) with zipfile.ZipFile(zip_file, 'r') as zip: diff --git a/src/hdmf/common/sparse.py b/src/hdmf/common/sparse.py index 0dd7d9654..8fe571018 100644 --- a/src/hdmf/common/sparse.py +++ b/src/hdmf/common/sparse.py @@ -8,27 +8,35 @@ class csr_matrix: # dummy class to prevent import errors from . import register_class from ..container import Container -from ..utils import docval, popargs, to_uint_array, get_data_shape, AllowPositional +from ..typing import ArrayData, validated +from ..utils import to_uint_array, get_data_shape, AllowPositional @register_class('CSRMatrix') class CSRMatrix(Container): - @docval({'name': 'data', 'type': (csr_matrix, 'array_data'), - 'doc': 'the data to use for this CSRMatrix or CSR data array.' - 'If passing CSR data array, *indices*, *indptr*, and *shape* must also be provided'}, - {'name': 'indices', 'type': 'array_data', 'doc': 'CSR index array', 'default': None}, - {'name': 'indptr', 'type': 'array_data', 'doc': 'CSR index pointer array', 'default': None}, - {'name': 'shape', 'type': 'array_data', 'doc': 'the shape of the matrix', 'default': None}, - {'name': 'name', 'type': str, 'doc': 'the name to use for this when storing', 'default': 'csr_matrix'}, - allow_positional=AllowPositional.WARNING) - def __init__(self, **kwargs): + @validated(allow_positional=AllowPositional.WARNING) + def __init__(self, + data: csr_matrix | ArrayData, + indices: ArrayData | None = None, + indptr: ArrayData | None = None, + shape: ArrayData | None = None, + name: str = 'csr_matrix'): + """Initialize the CSRMatrix. + + Args: + data: the data to use for this CSRMatrix or CSR data array. If passing CSR data array, + *indices*, *indptr*, and *shape* must also be provided + indices: CSR index array + indptr: CSR index pointer array + shape: the shape of the matrix + name: the name to use for this when storing + """ if not SCIPY_INSTALLED: raise ImportError( "scipy must be installed to use CSRMatrix. Please install scipy using `pip install scipy`." ) - data, indices, indptr, shape = popargs('data', 'indices', 'indptr', 'shape', kwargs) - super().__init__(**kwargs) + super().__init__(name=name) if not isinstance(data, csr_matrix): temp_shape = get_data_shape(data) temp_ndim = len(temp_shape) diff --git a/src/hdmf/data_utils.py b/src/hdmf/data_utils.py index 8f85378d6..6f09a5029 100644 --- a/src/hdmf/data_utils.py +++ b/src/hdmf/data_utils.py @@ -14,7 +14,10 @@ import h5py import numpy as np -from .utils import docval, getargs, popargs, docval_macro, get_data_shape, _is_collection, _get_length +from typing import Any + +from .typing import ArrayData, Bool, Float, Int, Shaped, validated +from .utils import docval_macro, get_data_shape, _is_collection, _get_length def append_data(data, arg): from hdmf.backends.hdf5.h5_utils import HDMFDataset @@ -156,61 +159,15 @@ def maxshape(self): class GenericDataChunkIterator(AbstractDataChunkIterator): """DataChunkIterator that lets the user specify chunk and buffer shapes.""" - __docval_init = ( - dict( - name="buffer_gb", - type=(float, int), - doc=( - "If buffer_shape is not specified, it will be inferred as the smallest chunk " - "below the buffer_gb threshold." - "Defaults to 1GB." - ), - default=None, - ), - dict( - name="buffer_shape", - type=tuple, - doc="Manually defined shape of the buffer.", - default=None, - ), - dict( - name="chunk_mb", - type=(float, int), - doc=( - "If chunk_shape is not specified, it will be inferred as the smallest chunk " - "below the chunk_mb threshold.", - "Defaults to 10MB.", - ), - default=None, - ), - dict( - name="chunk_shape", - type=tuple, - doc="Manually defined shape of the chunks.", - default=None, - ), - dict( - name="display_progress", - type=bool, - doc="Display a progress bar with iteration rate and estimated completion time.", - default=False, - ), - dict( - name="progress_bar_class", - type=Callable, - doc="The progress bar class to use. Defaults to tqdm.tqdm if the TQDM package is installed.", - default=None, - ), - dict( - name="progress_bar_options", - type=dict, - doc="Dictionary of keyword arguments to be passed directly to tqdm.", - default=None, - ), - ) - - @docval(*__docval_init) - def __init__(self, **kwargs): + @validated + def __init__(self, + buffer_gb: Float | Int | None = None, + buffer_shape: tuple | None = None, + chunk_mb: Float | Int | None = None, + chunk_shape: tuple | None = None, + display_progress: Bool = False, + progress_bar_class: Callable | None = None, + progress_bar_options: dict | None = None): """ Break a dataset into buffers containing multiple chunks to be written into an HDF5 dataset. @@ -220,25 +177,20 @@ def __init__(self, **kwargs): HDF5 recommends chunk size in the range of 2 to 16 MB for optimal cloud performance. https://youtu.be/rcS5vt-mKok?t=621 + + Args: + buffer_gb: If buffer_shape is not specified, it will be inferred as the smallest chunk + below the buffer_gb threshold. Defaults to 1GB. + buffer_shape: Manually defined shape of the buffer. + chunk_mb: If chunk_shape is not specified, it will be inferred as the smallest chunk + below the chunk_mb threshold. Defaults to 10MB. + chunk_shape: Manually defined shape of the chunks. + display_progress: Display a progress bar with iteration rate and estimated completion time. + progress_bar_class: The progress bar class to use. Defaults to tqdm.tqdm if the TQDM package + is installed. + progress_bar_options: Dictionary of keyword arguments to be passed directly to tqdm. """ - ( - buffer_gb, - buffer_shape, - chunk_mb, - chunk_shape, - self.display_progress, - progress_bar_class, - progress_bar_options, - ) = getargs( - "buffer_gb", - "buffer_shape", - "chunk_mb", - "chunk_shape", - "display_progress", - "progress_bar_class", - "progress_bar_options", - kwargs, - ) + self.display_progress = display_progress self.progress_bar_options = progress_bar_options or dict() if buffer_gb is None and buffer_shape is None: @@ -328,21 +280,15 @@ def __init__(self, **kwargs): ) self.display_progress = False - @docval( - dict( - name="chunk_mb", - type=(float, int), - doc="Size of the HDF5 chunk in megabytes.", - default=None, - ) - ) - def _get_default_chunk_shape(self, **kwargs) -> tuple[int, ...]: - """ - Select chunk shape with size in MB less than the threshold of chunk_mb. + @validated + def _get_default_chunk_shape(self, chunk_mb: Float | Int | None = None) -> tuple[int, ...]: + """Select chunk shape with size in MB less than the threshold of chunk_mb. Keeps the dimensional ratios of the original data. + + Args: + chunk_mb: Size of the HDF5 chunk in megabytes. """ - chunk_mb = getargs("chunk_mb", kwargs) assert chunk_mb > 0, f"chunk_mb ({chunk_mb}) must be greater than zero!" n_dims = len(self.maxshape) @@ -359,22 +305,16 @@ def _get_default_chunk_shape(self, **kwargs) -> tuple[int, ...]: k = math.floor((chunk_bytes / (prod_v * itemsize)) ** (1 / n_dims)) return tuple([min(k * x, self.maxshape[dim]) for dim, x in enumerate(v)]) - @docval( - dict( - name="buffer_gb", - type=(float, int), - doc="Size of the data buffer in gigabytes. Recommended to be as much free RAM as safely available.", - default=None, - ) - ) - def _get_default_buffer_shape(self, **kwargs) -> tuple[int, ...]: - """ - Select buffer shape with size in GB less than the threshold of buffer_gb. + @validated + def _get_default_buffer_shape(self, buffer_gb: Float | Int | None = None) -> tuple[int, ...]: + """Select buffer shape with size in GB less than the threshold of buffer_gb. Keeps the dimensional ratios of the original data. Assumes the chunk_shape has already been set. + + Args: + buffer_gb: Size of the data buffer in gigabytes. Recommended to be as much free RAM as safely available. """ - buffer_gb = getargs("buffer_gb", kwargs) assert buffer_gb > 0, f"buffer_gb ({buffer_gb}) must be greater than zero!" assert all(chunk_axis > 0 for chunk_axis in self.chunk_shape), ( f"Some dimensions of chunk_shape ({self.chunk_shape}) are less than zero!" @@ -496,29 +436,29 @@ class DataChunkIterator(AbstractDataChunkIterator): :py:class:`~hdmf.data_utils.AbstractDataChunkIterator` may be more appropriate. """ - __docval_init = ( - {'name': 'data', 'type': None, 'doc': 'The data object used for iteration', 'default': None}, - {'name': 'maxshape', 'type': tuple, - 'doc': 'The maximum shape of the full data array. Use None to indicate unlimited dimensions', - 'default': None}, - {'name': 'dtype', 'type': np.dtype, 'doc': 'The Numpy data type for the array', 'default': None}, - {'name': 'buffer_size', 'type': int, 'doc': 'Number of values to be buffered in a chunk', 'default': 1}, - {'name': 'iter_axis', 'type': int, 'doc': 'The dimension to iterate over', 'default': 0} - ) - - @docval(*__docval_init) - def __init__(self, **kwargs): + @validated + def __init__(self, + data: Any = None, + maxshape: tuple | None = None, + dtype: np.dtype | None = None, + buffer_size: Int = 1, + iter_axis: Int = 0): """Initialize the DataChunkIterator. If 'data' is an iterator and 'dtype' is not specified, then next is called on the iterator in order to determine the dtype of the data. + + Args: + data: The data object used for iteration + maxshape: The maximum shape of the full data array. Use None to indicate unlimited dimensions + dtype: The Numpy data type for the array + buffer_size: Number of values to be buffered in a chunk + iter_axis: The dimension to iterate over """ - # Get the user parameters - self.data, self.__maxshape, self.__dtype, self.buffer_size, self.iter_axis = getargs('data', - 'maxshape', - 'dtype', - 'buffer_size', - 'iter_axis', - kwargs) + self.data = data + self.__maxshape = maxshape + self.__dtype = dtype + self.buffer_size = buffer_size + self.iter_axis = iter_axis self.chunk_index = 0 # Create an iterator for the data if possible if isinstance(self.data, Iterable): @@ -569,9 +509,23 @@ def __init__(self, **kwargs): raise Exception('Data type could not be determined. Please specify dtype in DataChunkIterator init.') @classmethod - @docval(*__docval_init) - def from_iterable(cls, **kwargs): - return cls(**kwargs) + @validated + def from_iterable(cls, + data: Any = None, + maxshape: tuple | None = None, + dtype: np.dtype | None = None, + buffer_size: Int = 1, + iter_axis: Int = 0): + """Create a DataChunkIterator; takes the same arguments as the constructor. + + Args: + data: The data object used for iteration + maxshape: The maximum shape of the full data array. Use None to indicate unlimited dimensions + dtype: The Numpy data type for the array + buffer_size: Number of values to be buffered in a chunk + iter_axis: The dimension to iterate over + """ + return cls(data=data, maxshape=maxshape, dtype=dtype, buffer_size=buffer_size, iter_axis=iter_axis) def __iter__(self): """Return the iterator object""" @@ -686,22 +640,28 @@ def __next__(self): next = __next__ - @docval(returns='Tuple with the recommended chunk shape or None if no particular shape is recommended.') def recommended_chunk_shape(self): """Recommend a chunk shape. To optimize iterative write the chunk should be aligned with the common shape of chunks returned by __next__ or if those chunks are too large, then a well-aligned subset of those chunks. This may also be any other value in case one wants to recommend chunk shapes to optimize read rather - than write. The default implementation returns None, indicating no preferential chunking option.""" + than write. The default implementation returns None, indicating no preferential chunking option. + + Returns: + Tuple with the recommended chunk shape or None if no particular shape is recommended. + """ return None - @docval(returns='Recommended initial shape for the full data. This should be the shape of the full dataset' + - 'if known beforehand or alternatively the minimum shape of the dataset. Return None if no ' + - 'recommendation is available') def recommended_data_shape(self): """Recommend an initial shape of the data. This is useful when progressively writing data and - we want to recommend an initial size for the dataset""" + we want to recommend an initial size for the dataset. + + Returns: + Recommended initial shape for the full data. This should be the shape of the full dataset + if known beforehand or alternatively the minimum shape of the dataset. Return None if no + recommendation is available. + """ if self.maxshape is not None: if np.all([i is not None for i in self.maxshape]): return self.maxshape @@ -764,12 +724,15 @@ class DataChunk: Class used to describe a data chunk. Used in DataChunkIterator. """ - @docval({'name': 'data', 'type': np.ndarray, - 'doc': 'Numpy array with the data value(s) of the chunk', 'default': None}, - {'name': 'selection', 'type': None, - 'doc': 'Numpy index tuple describing the location of the chunk', 'default': None}) - def __init__(self, **kwargs): - self.data, self.selection = getargs('data', 'selection', kwargs) + @validated + def __init__(self, data: np.ndarray | None = None, selection: Any = None): + """Initialize the DataChunk. + + Args: + data: Numpy array with the data value(s) of the chunk + selection: Numpy index tuple describing the location of the chunk + """ + self.data, self.selection = data, selection def __len__(self): """Get the number of values in the data chunk""" @@ -954,27 +917,39 @@ class ShapeValidatorResult: values are strings with default error messages for the type. """ - @docval({'name': 'result', 'type': bool, 'doc': 'Result of the shape validation', 'default': False}, - {'name': 'message', 'type': str, - 'doc': 'Message describing the result of the shape validation', 'default': None}, - {'name': 'ignored', 'type': tuple, - 'doc': 'Axes that have been ignored in the validation process', 'default': tuple(), 'shape': (None,)}, - {'name': 'unmatched', 'type': tuple, - 'doc': 'List of axes that did not match during shape validation', 'default': tuple(), 'shape': (None,)}, - {'name': 'error', 'type': str, 'doc': 'Error that may have occurred. One of ERROR_TYPE', 'default': None}, - {'name': 'shape1', 'type': tuple, - 'doc': 'Shape of the first array for comparison', 'default': tuple(), 'shape': (None,)}, - {'name': 'shape2', 'type': tuple, - 'doc': 'Shape of the second array for comparison', 'default': tuple(), 'shape': (None,)}, - {'name': 'axes1', 'type': tuple, - 'doc': 'Axes for the first array that should match', 'default': tuple(), 'shape': (None,)}, - {'name': 'axes2', 'type': tuple, - 'doc': 'Axes for the second array that should match', 'default': tuple(), 'shape': (None,)}, - ) - def __init__(self, **kwargs): - self.result, self.message, self.ignored, self.unmatched, \ - self.error, self.shape1, self.shape2, self.axes1, self.axes2 = getargs( - 'result', 'message', 'ignored', 'unmatched', 'error', 'shape1', 'shape2', 'axes1', 'axes2', kwargs) + @validated + def __init__(self, + result: Bool = False, + message: str | None = None, + ignored: Shaped[tuple, (None,)] = (), + unmatched: Shaped[tuple, (None,)] = (), + error: str | None = None, + shape1: Shaped[tuple, (None,)] = (), + shape2: Shaped[tuple, (None,)] = (), + axes1: Shaped[tuple, (None,)] = (), + axes2: Shaped[tuple, (None,)] = ()): + """Initialize the ShapeValidatorResult. + + Args: + result: Result of the shape validation + message: Message describing the result of the shape validation + ignored: Axes that have been ignored in the validation process + unmatched: List of axes that did not match during shape validation + error: Error that may have occurred. One of ERROR_TYPE + shape1: Shape of the first array for comparison + shape2: Shape of the second array for comparison + axes1: Axes for the first array that should match + axes2: Axes for the second array that should match + """ + self.result = result + self.message = message + self.ignored = ignored + self.unmatched = unmatched + self.error = error + self.shape1 = shape1 + self.shape2 = shape2 + self.axes1 = axes1 + self.axes2 = axes2 def __setattr__(self, key, value): """ @@ -1007,20 +982,18 @@ class DataIO: used to pass dataset-specific I/O parameters to the particular HDMFIO backend. """ - @docval({'name': 'data', - 'type': 'array_data', - 'doc': 'the data to be written', - 'default': None}, - {'name': 'dtype', - 'type': (type, np.dtype), - 'doc': 'the data type of the dataset. Not used if data is specified.', - 'default': None}, - {'name': 'shape', - 'type': tuple, - 'doc': 'the shape of the dataset. Not used if data is specified.', - 'default': None}) - def __init__(self, **kwargs): - data, dtype, shape = popargs('data', 'dtype', 'shape', kwargs) + @validated + def __init__(self, + data: ArrayData | None = None, + dtype: type | np.dtype | None = None, + shape: tuple | None = None): + """Initialize the DataIO, wrapping data or declaring a dtype and shape for later. + + Args: + data: the data to be written + dtype: the data type of the dataset. Not used if data is specified. + shape: the shape of the dataset. Not used if data is specified. + """ if data is None: if (dtype is None) ^ (shape is None): raise ValueError("Must specify 'dtype' and 'shape' if not specifying 'data'") diff --git a/src/hdmf/query.py b/src/hdmf/query.py index abe2a93a7..ca45adc5b 100644 --- a/src/hdmf/query.py +++ b/src/hdmf/query.py @@ -2,7 +2,8 @@ import numpy as np -from .utils import ExtenderMeta, docval_macro, docval, getargs +from .typing import ArrayData, validated +from .utils import ExtenderMeta, docval_macro @docval_macro('array_data') @@ -19,10 +20,15 @@ def __getitem__(self, key): idx = self.__evaluate_key(key) return self.dataset[idx] - @docval({'name': 'dataset', 'type': 'array_data', 'doc': 'the HDF5 file lazily evaluate'}) - def __init__(self, **kwargs): + @validated + def __init__(self, dataset: ArrayData): + """Initialize the HDMFDataset. + + Args: + dataset: the dataset (e.g. an HDF5 or Zarr dataset) to be lazily evaluated + """ super().__init__() - self.__dataset = getargs('dataset', kwargs) + self.__dataset = dataset @property def dataset(self): diff --git a/src/hdmf/typing/__init__.py b/src/hdmf/typing/__init__.py new file mode 100644 index 000000000..e9444ee7d --- /dev/null +++ b/src/hdmf/typing/__init__.py @@ -0,0 +1,59 @@ +"""Type hints and runtime validation for HDMF (successor to ``@docval``). + +This package provides: + +- Type aliases enforceable by beartype anywhere (``@beartype``, + ``beartype.door.is_bearable``, or HDMF's :func:`validated`): numeric aliases + (``Int``, ``UInt``, ``Float``, ``Bool``) accepting numpy scalar + types, macro aliases (``ArrayData``, ``ScalarData``, ``AnyData``) + backed by a live type registry, :class:`TypeName` for cross-module forward + references, and :class:`Shaped` for array shape requirements. numpydantic + ``NDArray[...]`` hints are also supported for dtype- and shape-checked arrays. +- The :func:`validated` decorator, which validates arguments of a type-hinted + function at call time through beartype/numpydantic. +- A compatibility layer so :func:`hdmf.utils.get_docval` works on plain type-hinted + functions: docval-format argument specs are synthesized from the signature, type + hints, and Google-style docstring. This keeps downstream code that splices parent + argument specs (``@docval(*get_docval(Parent.__init__, ...))``) working while + ``@docval`` is phased out (https://github.com/hdmf-dev/hdmf/issues/1129); it is + the only part of this package that speaks docval, and it will be removed together + with docval. + +Migration tooling lives in :mod:`hdmf.typing.migrate` +(``python -m hdmf.typing.migrate --help``) and parity-testing helpers in +:mod:`hdmf.typing.testing`. +""" + +from ..utils import AllowPositional # re-export: used as a @validated option +from ._compat import map_hint, synthesize_docval +from ._decorator import set_type_checking, validated +from ._shapes import Shaped +from ._types import ( + AnyData, + ArrayData, + Bool, + Float, + Int, + ScalarData, + TypeName, + UInt, + register_macro, +) + +__all__ = [ + 'AllowPositional', + 'AnyData', + 'ArrayData', + 'Bool', + 'Float', + 'Int', + 'ScalarData', + 'Shaped', + 'TypeName', + 'UInt', + 'map_hint', + 'register_macro', + 'set_type_checking', + 'synthesize_docval', + 'validated', +] diff --git a/src/hdmf/typing/_compat.py b/src/hdmf/typing/_compat.py new file mode 100644 index 000000000..0e55efb8e --- /dev/null +++ b/src/hdmf/typing/_compat.py @@ -0,0 +1,286 @@ +"""Synthesis of docval argument specifications from type-hinted signatures. + +This is the compatibility linchpin for the migration away from ``@docval`` +(see https://github.com/hdmf-dev/hdmf/issues/1129): :func:`hdmf.utils.get_docval` +falls back to :func:`synthesize_docval` for functions that have type hints instead +of a ``__docval__`` attribute. The synthesized dicts use only keys and value forms +that legacy ``@docval`` accepts, so they can be spliced into downstream decorators +(``@docval(*get_docval(Parent.__init__, ...))``) and consumed by HDMF's build +machinery unchanged. +""" + +import inspect +import types +import typing +from typing import Annotated, Any, Literal + +from ._docstrings import parse_docstring +from ._types import _bool_types, _float_types, _int_types, _uint_types +from ._validators import compat_info + +_synth_attr_name = '__synth_docval__' + +_NoneType = type(None) + +# used to render numeric alias unions back to docval's numeric vocabulary; order +# matters only for readability of the resulting spec +_NUMERIC_ALIASES = ( + ('uint', frozenset(_uint_types)), + ('int', frozenset(_int_types)), + ('float', frozenset(_float_types)), + ('bool', frozenset(_bool_types)), +) + + +def _collapse_numeric(members): + """Replace complete numeric-alias type sets in a union with docval's string names. + + E.g. ``(int, np.int8, ..., str)`` (from an ``Int | str`` hint) becomes + ``('int', str)``. + """ + mset = {m for m in members if isinstance(m, type)} + member_to_name = {} + for name, type_set in _NUMERIC_ALIASES: + if type_set <= mset: + for t in type_set: + member_to_name[t] = name + out = [] + for m in members: + name = member_to_name.get(m) if isinstance(m, type) else None + if name is None: + out.append(m) + elif name not in out: + out.append(name) + return tuple(out) + + +class MappedHint: + """Result of mapping one type hint to docval spec fields. + + ``fields`` holds the docval spec keys derived from the hint ('type', and + optionally 'shape' and 'enum'). ``exact`` is False when the mapping lost + information (e.g. ``list[int]`` degraded to ``list``), in which case runtime + validation should prefer the original hint over the synthesized spec. + ``none_allowed`` is True when the hint had an explicit ``| None`` member. + """ + + __slots__ = ('fields', 'exact', 'none_allowed') + + def __init__(self, fields, exact=True, none_allowed=False): + self.fields = fields + self.exact = exact + self.none_allowed = none_allowed + + +def _is_numpydantic_ndarray(hint): + """Return True if the hint is a numpydantic/nptyping NDArray specialization.""" + mod = getattr(type(hint), '__module__', '') or '' + if not (mod.startswith('numpydantic') or mod.startswith('nptyping')): + return False + return 'NDArray' in (getattr(hint, '__name__', '') or repr(hint)) + + +def _ndarray_shape_tuple(hint): + """Best-effort extraction of a docval shape tuple from an NDArray hint. + + Returns None when the shape is unconstrained or cannot be interpreted. + """ + try: + shape_expr = hint.__args__[0] # nptyping-style: NDArray[Shape[...], dtype] + entries = getattr(shape_expr, 'prepared_args', None) + if not entries: + return None + dims = [] + for entry in entries: + # entries look like '*', '2', '* x', '3 y' + size = str(entry).split()[0] + if size == '*': + dims.append(None) + elif size.isdigit(): + dims.append(int(size)) + else: + return None # named/variadic dims we cannot express in docval + return tuple(dims) + except Exception: + return None + + +def map_hint(hint): # noqa: C901 + """Map a single type hint to docval spec fields. Returns a ``MappedHint``.""" + if hint is None or hint is _NoneType or hint is inspect.Parameter.empty or hint is Any: + return MappedHint({'type': None}) + + # unresolved forward reference: docval matches the string against the MRO + if isinstance(hint, str): + return MappedHint({'type': hint}) + if isinstance(hint, typing.ForwardRef): + return MappedHint({'type': hint.__forward_arg__}) + + origin = typing.get_origin(hint) + + if origin is Annotated: + base, *metadata = typing.get_args(hint) + docval_name = None + shape = None + for meta in metadata: + info = compat_info(meta) + if info is None: + continue + if 'docval_name' in info: + docval_name = info['docval_name'] + if 'shape' in info: + shape = info['shape'] + if docval_name is not None: + fields = {'type': docval_name} + if shape is not None: + fields['shape'] = shape + return MappedHint(fields) + mapped = map_hint(base) + if shape is not None: + mapped.fields['shape'] = shape + return mapped + + if origin in (typing.Union, types.UnionType): + members = [] + exact = True + none_allowed = False + shape = None + enum = None + for arg in typing.get_args(hint): + if arg is _NoneType: + none_allowed = True + continue + mapped = map_hint(arg) + exact = exact and mapped.exact + shape = shape if shape is not None else mapped.fields.get('shape') + enum = enum if enum is not None else mapped.fields.get('enum') + member_type = mapped.fields['type'] + if isinstance(member_type, tuple): + members.extend(member_type) + else: + members.append(member_type) + # dedupe while preserving order (unhashable members are not produced here), + # then render complete numeric alias sets back to docval's string names + members = _collapse_numeric(tuple(dict.fromkeys(members))) + fields = {'type': members[0] if len(members) == 1 else members} + if shape is not None: + fields['shape'] = shape + if enum is not None: + fields['enum'] = enum + return MappedHint(fields, exact=exact, none_allowed=none_allowed) + + if origin is Literal: + values = typing.get_args(hint) + value_types = tuple(dict.fromkeys(type(v) for v in values)) + fields = { + 'type': value_types[0] if len(value_types) == 1 else value_types, + 'enum': tuple(values), + } + return MappedHint(fields) + + if _is_numpydantic_ndarray(hint): + fields = {'type': 'array_data'} + shape = _ndarray_shape_tuple(hint) + if shape is not None: + fields['shape'] = shape + # numpydantic's isinstance() checks dtype and named dims that docval cannot; + # runtime validation should use the hint itself + return MappedHint(fields, exact=False) + + if origin is not None: + # parametrized generic (list[int], dict[str, int], Callable[..., X], ...): + # docval never checked element types, so degrade to the bare origin class + if isinstance(origin, type): + return MappedHint({'type': origin}, exact=False) + return MappedHint({'type': None}, exact=False) + + if isinstance(hint, type): + return MappedHint({'type': hint}) + + # anything else (TypeVar, Protocol instance, special form): docval "any type" + return MappedHint({'type': None}, exact=False) + + +def _safe_hints(func): + """Return the function's type hints with extras, tolerating unresolvable names. + + ``typing.get_type_hints`` raises on the first unresolvable forward reference; in + that case fall back to the raw ``__annotations__``, evaluating each entry + individually and keeping unresolvable ones as strings (docval MRO-name semantics). + """ + try: + return typing.get_type_hints(func, include_extras=True) + except Exception: + pass + hints = {} + raw = getattr(func, '__annotations__', {}) + globalns = getattr(func, '__globals__', {}) + for name, annotation in raw.items(): + if isinstance(annotation, str): + try: + hints[name] = eval(annotation, globalns) # noqa: S307 + except Exception: + hints[name] = annotation + else: + hints[name] = annotation + return hints + + +def synthesize_docval(func): + """Synthesize docval argument specs from a type-hinted function. + + Returns ``(specs, idx, meta)`` where ``specs`` is a tuple of docval-compatible + argument spec dicts in signature order, ``idx`` maps argument name to spec, and + ``meta`` is a dict with function-level info: 'allow_extra' (function accepts + ``**kwargs``), 'rtype'/'returns' (from the return annotation and docstring), and + 'exact' (name -> bool, False where the spec is a lossy rendering of the hint). + The result is cached on the function. + """ + target = inspect.unwrap(getattr(func, '__func__', func)) + cached = getattr(target, _synth_attr_name, None) + if cached is not None: + return cached + + sig = inspect.signature(target) + hints = _safe_hints(target) + param_docs, returns_doc, _ = parse_docstring(target) + + specs = [] + exact = {} + allow_extra = False + for name, param in sig.parameters.items(): + if name in ('self', 'cls'): + continue + if param.kind is inspect.Parameter.VAR_KEYWORD: + allow_extra = True + continue + if param.kind is inspect.Parameter.VAR_POSITIONAL: + raise TypeError( + f"cannot synthesize docval arguments for {target.__qualname__}: " + "variadic positional arguments (*args) are not supported" + ) + mapped = map_hint(hints.get(name, param.annotation)) + spec = {'name': name, 'doc': param_docs.get(name, '')} + spec.update(mapped.fields) + if param.default is not inspect.Parameter.empty: + spec['default'] = param.default + if mapped.none_allowed and param.default is not None: + spec['allow_none'] = True + # required args with `| None` hints are not expressible in docval and the + # spec omits None; see the migration guide + specs.append(spec) + exact[name] = mapped.exact + + rtype_hint = hints.get('return') + meta = { + 'allow_extra': allow_extra, + 'rtype': rtype_hint if rtype_hint is not None else None, + 'returns': returns_doc, + 'exact': exact, + } + result = (tuple(specs), {s['name']: s for s in specs}, meta) + try: + setattr(target, _synth_attr_name, result) + except (AttributeError, TypeError): # non-writable callables (builtins, slots) + pass + return result diff --git a/src/hdmf/typing/_decorator.py b/src/hdmf/typing/_decorator.py new file mode 100644 index 000000000..7a173e8f9 --- /dev/null +++ b/src/hdmf/typing/_decorator.py @@ -0,0 +1,231 @@ +"""The @validated decorator: runtime validation driven purely by type hints. + +Validation runs through beartype (``beartype.door.is_bearable``) for every +parameter, including the :mod:`hdmf.typing` aliases, whose semantics are +implemented as beartype validators (see ``_validators``). numpydantic ``NDArray`` +hints are checked through numpydantic's own ``isinstance`` machinery. No docval +machinery is involved; the docval-format specs synthesized at decoration time exist +only so :func:`hdmf.utils.get_docval` keeps working for downstream code during the +migration, and are otherwise used here only to phrase error messages. +""" + +import functools +import inspect +import os +import typing +import warnings + +from beartype.door import is_bearable + +from ..utils import AllowPositional +from ._compat import _is_numpydantic_ndarray, _safe_hints, synthesize_docval +from ._shapes import check_shape +from ._validators import compat_info, matches_type_name + +_TYPE_CHECKING_ENABLED = os.environ.get('HDMF_TYPE_CHECKING', '').lower() not in ('off', '0', 'false') + + +def set_type_checking(enabled): + """Globally enable or disable call-time validation by @validated functions. + + Validation can also be disabled by setting the environment variable + ``HDMF_TYPE_CHECKING=off`` before importing hdmf. + """ + global _TYPE_CHECKING_ENABLED + _TYPE_CHECKING_ENABLED = bool(enabled) + + +def _format_type(argtype): + # renders synthesized docval-vocabulary types into readable error messages + if isinstance(argtype, str): + return argtype + elif isinstance(argtype, type): + return argtype.__name__ + elif isinstance(argtype, (tuple, list)): + parts = [_format_type(i) for i in argtype] + if len(parts) > 1: + return "%s or %s" % (", ".join(parts[:-1]), parts[-1]) + return parts[0] + elif argtype is None: + return "any type" + raise ValueError("argtype must be a type, str, list, or tuple") + + +def _fmt_str_quotes(x): + if isinstance(x, (list, tuple)): + return '{}'.format(x) + if isinstance(x, str): + return "'%s'" % x + return str(x) + + +def _without_shape_validators(hint): + """Return the hint with HDMF shape validators removed. + + ``@validated`` runs its own shape pass (with the unwrap-by-argument-name + fallback and shape violations raising ValueError, matching docval), so shape + must not also fail the beartype type check, where it would be misclassified as + a type error. The full annotation still enforces shape for plain-beartype + consumers. + """ + if typing.get_origin(hint) is not typing.Annotated: + return hint + base, *metadata = typing.get_args(hint) + kept = [m for m in metadata if 'shape' not in (compat_info(m) or {})] + if len(kept) == len(metadata): + return hint + if kept: + return typing.Annotated[tuple([base] + kept)] + return base + + +class _ParamCheck: + """Precomputed validation info for one parameter.""" + + __slots__ = ('name', 'spec', 'hint', 'checker', 'allow_none', 'required') + + def __init__(self, name, spec, hint, func_qualname): + self.name = name + self.spec = spec + self.hint = hint + self.required = 'default' not in spec + self.allow_none = not self.required and (spec['default'] is None or spec.get('allow_none', False)) + self.checker = self._build_checker(func_qualname) + + def _build_checker(self, func_qualname): + hint = _without_shape_validators(self.hint) + if hint is None or hint is inspect.Parameter.empty or hint is object: + return None # unannotated: accept anything + if 'enum' in self.spec: + # Literal membership is checked separately (ValueError, like docval); + # here only check the value's type against the literal values' types + base_types = tuple(dict.fromkeys(type(v) for v in self.spec['enum'])) + return lambda value: isinstance(value, base_types) + if isinstance(hint, str): + # unresolvable forward reference: match by class name in the MRO + return lambda value: matches_type_name(value, hint) + if isinstance(hint, typing.ForwardRef): + forward_name = hint.__forward_arg__ + return lambda value: matches_type_name(value, forward_name) + if _is_numpydantic_ndarray(hint): + # numpydantic implements isinstance() with dtype and shape checking + return lambda value: isinstance(value, hint) + try: + is_bearable(None, hint) # force beartype to compile the hint now + except Exception as e: + warnings.warn( + f"{func_qualname}: type hint {hint!r} for argument '{self.name}' is not " + f"checkable at runtime ({type(e).__name__}); it will not be validated", + stacklevel=4, + ) + return None + return lambda value: is_bearable(value, hint) + + def type_ok(self, argval): + if argval is None: + # match the long-standing `arg: T = None` idiom: None is valid whenever + # the default is None, even if the hint does not include None + return self.allow_none or (self.checker is not None and self.checker(None)) + return self.checker is None or self.checker(argval) + + def expected_str(self): + spec_type = self.spec.get('type') + if spec_type is not None: + return _format_type(spec_type) + return str(self.hint) + + +def validated(func=None, *, enforce_type=True, enforce_shape=True, + allow_positional=AllowPositional.ALLOWED): + """Decorator validating arguments of a type-hinted function at call time. + + Args: + enforce_type: whether to check argument types at call time + enforce_shape: whether to check array shapes (from ``Shaped[...]`` hints) + allow_positional: policy for positional arguments. Prefer real keyword-only + parameters (``*,``) over ``AllowPositional.ERROR`` in new code. + """ + if func is None: + return lambda f: _apply_validated(f, enforce_type, enforce_shape, allow_positional) + return _apply_validated(func, enforce_type, enforce_shape, allow_positional) + + +def _apply_validated(func, enforce_type, enforce_shape, allow_positional): # noqa: C901 + sig = inspect.signature(func) + specs, idx, meta = synthesize_docval(func) + hints = _safe_hints(func) + + param_names = list(sig.parameters) + has_receiver = bool(param_names) and param_names[0] in ('self', 'cls') + + checks = {} + for spec in specs: + name = spec['name'] + hint = hints.get(name, sig.parameters[name].annotation) + checks[name] = _ParamCheck(name, spec, hint, func.__qualname__) + + @functools.wraps(func) + def wrapper(*args, **kwargs): + if not _TYPE_CHECKING_ENABLED: + return func(*args, **kwargs) + + n_positional = len(args) - (1 if has_receiver else 0) + if n_positional > 0: + if allow_positional == AllowPositional.WARNING: + msg = ('%s: Using positional arguments for this method is discouraged and will be ' + 'deprecated in a future major release. Please use keyword arguments to ' + 'ensure future compatibility.' % func.__qualname__) + warnings.warn(msg, category=FutureWarning, stacklevel=2) + elif allow_positional == AllowPositional.ERROR: + msg = ('%s: Only keyword arguments (e.g., func(argname=value, ...)) are allowed ' + 'for this method.' % func.__qualname__) + raise SyntaxError(msg) + + try: + bound = sig.bind(*args, **kwargs) + except TypeError as e: + raise TypeError('%s: %s' % (func.__qualname__, e)) from None + + type_errors = [] + value_errors = [] + from ..term_set import TermSetWrapper # deferred to avoid a circular import + + for name, argval in bound.arguments.items(): + check = checks.get(name) + if check is None: # self/cls or **kwargs extras: passed through unvalidated + continue + if isinstance(argval, TermSetWrapper): + # validate the wrapped value; the wrapper itself is passed to the body + argval = argval.value + spec = check.spec + if enforce_type: + if not check.type_ok(argval): + if argval is None and (check.required or spec['default'] is None): + type_errors.append("None is not allowed for '%s' (expected '%s', not None)" + % (name, check.expected_str())) + else: + type_errors.append("incorrect type for '%s' (got '%s', expected '%s')" + % (name, type(argval).__name__, check.expected_str())) + if enforce_shape and 'shape' in spec and argval is not None: + err = check_shape(name, argval, spec['shape']) + if err is not None: + value_errors.append(err) + if 'enum' in spec and argval is not None and argval not in spec['enum']: + value_errors.append("forbidden value for '%s' (got %s, expected %s)" + % (name, _fmt_str_quotes(argval), spec['enum'])) + + if type_errors: + raise TypeError('%s: %s' % (func.__qualname__, ', '.join(type_errors))) + if value_errors: + raise ValueError('%s: %s' % (func.__qualname__, ', '.join(value_errors))) + + return func(*args, **kwargs) + + wrapper.__validated__ = { + 'args': specs, + 'enforce_type': enforce_type, + 'enforce_shape': enforce_shape, + 'allow_positional': allow_positional, + 'allow_extra': meta['allow_extra'], + } + return wrapper diff --git a/src/hdmf/typing/_docstrings.py b/src/hdmf/typing/_docstrings.py new file mode 100644 index 000000000..587dadf05 --- /dev/null +++ b/src/hdmf/typing/_docstrings.py @@ -0,0 +1,31 @@ +"""Extraction of per-argument documentation from Google-style docstrings.""" + +from docstring_parser import DocstringStyle, parse + + +def parse_docstring(func): + """Return (param_docs, returns_doc, rtype) parsed from ``func.__doc__``. + + ``param_docs`` maps argument name to its description from the ``Args:`` section + (missing arguments simply have no entry). ``returns_doc`` and ``rtype`` come from + the ``Returns:`` section, or are None if absent. + """ + doc = getattr(func, '__doc__', None) + if not doc: + return {}, None, None + try: + parsed = parse(doc, style=DocstringStyle.GOOGLE) + except Exception: + return {}, None, None + param_docs = {} + for param in parsed.params: + if param.description: + # collapse continuation-line whitespace into single spaces + param_docs[param.arg_name] = ' '.join(param.description.split()) + returns_doc = None + rtype = None + if parsed.returns is not None: + if parsed.returns.description: + returns_doc = ' '.join(parsed.returns.description.split()) + rtype = parsed.returns.type_name + return param_docs, returns_doc, rtype diff --git a/src/hdmf/typing/_shapes.py b/src/hdmf/typing/_shapes.py new file mode 100644 index 000000000..f4a849f5a --- /dev/null +++ b/src/hdmf/typing/_shapes.py @@ -0,0 +1,77 @@ +"""Array shape annotations and validation. + +For plain array arguments, prefer numpydantic (``NDArray[Shape["* x, 3 y"], ...]``), +which checks dtype and shape through its own ``isinstance`` machinery. ``Shaped`` +exists for HDMF's looser shape semantics: it accepts anything (not just arrays), and +under ``@validated`` it applies the historical fallback of unwrapping a value by +argument name (e.g. a ``TimeSeries`` passed for an argument named ``data`` has its +``.data`` checked) when the value's own shape cannot be determined. +""" + +from typing import Annotated + +from ._validators import shape_validator + + +class Shaped: + """Annotate a type with a required array shape: ``Shaped[ArrayData, (None, 3)]``. + + A shape is a tuple where each element is an int (exact dimension length) or None + (any length); a tuple of such tuples means any of the listed shapes is allowed. + The annotation is enforced by beartype wherever the value's shape is + determinable; ``@validated`` additionally applies the unwrap-by-argument-name + fallback. + """ + + def __class_getitem__(cls, item): + if not (isinstance(item, tuple) and len(item) == 2): + raise TypeError("Shaped[...] requires two arguments: Shaped[type, shape]") + t, shape = item + if not isinstance(shape, (tuple, list)): + raise TypeError(f"shape must be a tuple or list, got {shape!r}") + shape = tuple(tuple(s) if isinstance(s, (tuple, list)) else s for s in shape) + return Annotated[t, shape_validator(shape)] + + +def _shape_okay(valshape, argshape): + if len(valshape) != len(argshape): + return False + return all(b in (a, None) for a, b in zip(valshape, argshape)) + + +def _shape_okay_multi(valshape, argshape): + if argshape and isinstance(argshape[0], (tuple, list)): # multiple allowable shapes + return any(_shape_okay(valshape, a) for a in argshape) + return _shape_okay(valshape, argshape) + + +def _shape_error_message(argname, valshape, allowable_shapes): + if isinstance(allowable_shapes, (list, tuple)) and all(isinstance(e, (list, tuple)) for e in allowable_shapes): + allowable_shapes_str = " or ".join(map(str, allowable_shapes)) + else: + allowable_shapes_str = str(allowable_shapes) + allowable_shapes_str = allowable_shapes_str.replace("None", "*") + return f"incorrect shape for {argname}: got {valshape}, and expected {allowable_shapes_str}" + + +def check_shape(argname, value, shape): + """Check ``value`` against a shape spec, unwrapping by argument name if needed. + + Returns None if the shape validates, otherwise an error message string. + """ + from ..utils import get_data_shape + + argval = value + valshape = get_data_shape(argval) + while valshape is None: + if argval is None: + return None + if not hasattr(argval, argname): + return ("cannot check shape of object '%s' for argument '%s' (expected shape '%s')" + % (argval, argname, shape)) + # unpack, e.g. if TimeSeries is passed for arg 'data', then TimeSeries.data is checked + argval = getattr(argval, argname) + valshape = get_data_shape(argval) + if valshape is not None and not _shape_okay_multi(valshape, shape): + return _shape_error_message(argname, valshape, shape) + return None diff --git a/src/hdmf/typing/_types.py b/src/hdmf/typing/_types.py new file mode 100644 index 000000000..ce251a52b --- /dev/null +++ b/src/hdmf/typing/_types.py @@ -0,0 +1,67 @@ +"""Type aliases for use in type-hinted HDMF function signatures. + +All aliases are enforceable by beartype directly (``@beartype``, +``beartype.door.is_bearable``, or HDMF's :func:`~hdmf.typing.validated`): the +numeric aliases are plain unions, and the macro/name aliases carry +:mod:`beartype.vale` validators. The ``get_docval`` compatibility shim knows how to +render each alias back into legacy docval vocabulary for downstream code that still +splices docval specs; that is the aliases' only connection to docval. +""" + +import typing +from typing import Annotated, Any + +import numpy as np + +from ._validators import macro_validator, type_name_validator + +# numeric aliases accepting numpy scalar types alongside the Python types, matching +# how HDMF has always treated numeric data. A bare `int` hint is strict under +# beartype and will NOT accept np.int32 — use these aliases for numeric arguments. +_int_types = (int, np.int8, np.int16, np.int32, np.int64) +_uint_types = (np.uint8, np.uint16, np.uint32, np.uint64) +_float_types = [float, np.float16, np.float32, np.float64] +if hasattr(np, "float128"): # pragma: no cover + _float_types.append(np.float128) +if hasattr(np, "longdouble"): # pragma: no cover + _float_types.append(np.longdouble) +_float_types = tuple(dict.fromkeys(_float_types)) # dedupe (longdouble may alias float128) +_bool_types = (bool, np.bool_) + +Int = typing.Union[_int_types] +UInt = typing.Union[_uint_types] +Float = typing.Union[_float_types] +Bool = typing.Union[_bool_types] + +# macro aliases accept instances of any type registered under the corresponding +# macro name; the registry is read at call time, so types registered later +# (e.g. by hdmf-zarr) are honored +ArrayData = Annotated[Any, macro_validator('array_data')] +ScalarData = Annotated[Any, macro_validator('scalar_data')] +AnyData = Annotated[Any, macro_validator('data')] + + +class TypeName: + """Reference a type by class name, matched against the value's MRO at call time. + + ``TypeName['DynamicTable']`` accepts any value with a class named + ``DynamicTable`` (or with that fully qualified ``module.qualname``) anywhere in + its MRO. Use this for forward references that cross module boundaries, where a + PEP 484 string annotation would not resolve. + """ + + def __class_getitem__(cls, name): + if not isinstance(name, str): + raise TypeError(f"TypeName[...] requires a class name string, got {name!r}") + return Annotated[Any, type_name_validator(name)] + + +def register_macro(macro_name): + """Class decorator registering a type under a macro name (e.g. ``'array_data'``). + + Successor to :func:`hdmf.utils.docval_macro`; both write to the same registry, + which the macro aliases (``ArrayData``, ``ScalarData``, ``AnyData``) + read at call time. + """ + from ..utils import docval_macro + return docval_macro(macro_name) diff --git a/src/hdmf/typing/_validators.py b/src/hdmf/typing/_validators.py new file mode 100644 index 000000000..acb91ade9 --- /dev/null +++ b/src/hdmf/typing/_validators.py @@ -0,0 +1,112 @@ +"""beartype validators implementing HDMF's type semantics. + +These are real :mod:`beartype.vale` validators, so any beartype-aware consumer +(``@beartype``, ``beartype.door.is_bearable``, or HDMF's ``@validated``) enforces +them natively. Each validator also carries compatibility info in a side registry +(see :func:`compat_info`) so the ``get_docval`` shim can render it back into legacy +docval vocabulary; that shim is the only docval-facing piece, and it goes away when +docval does. +""" + +from beartype.vale import Is + +# side registry: beartype validators are slotted and cannot carry attributes, so map +# validator identity -> compat info (with a strong reference to keep the id stable) +_COMPAT_INFO = {} + + +def _remember(validator, info): + _COMPAT_INFO[id(validator)] = (validator, info) + return validator + + +def compat_info(metadata): + """Return the compat-info dict for an HDMF validator, or None for foreign metadata.""" + entry = _COMPAT_INFO.get(id(metadata)) + return entry[1] if entry is not None else None + + +def matches_type_name(value, name): + """Return True if any class in the value's MRO matches ``name``. + + ``name`` may be a bare class name or a fully qualified ``module.qualname``. + This mirrors how docval matched string type names, without importing the type. + """ + for cls in type(value).__mro__: + if cls.__name__ == name or f"{cls.__module__}.{cls.__qualname__}" == name: + return True + return False + + +_macro_validators = {} +_type_name_validators = {} +_shape_validators = {} + + +def macro_validator(macro_name): + """A validator accepting instances of any type registered under ``macro_name``. + + The registry is read at call time, so types registered later (e.g. by hdmf-zarr) + are honored. + """ + if macro_name in _macro_validators: + return _macro_validators[macro_name] + + def checker(value): + types_ = _macro_types(macro_name) + return bool(types_) and isinstance(value, types_) + + checker.__name__ = checker.__qualname__ = f"is_{macro_name}" + validator = _remember(Is[checker], {'docval_name': macro_name}) + _macro_validators[macro_name] = validator + return validator + + +def _macro_types(macro_name): + # the registry currently lives in hdmf.utils for docval compatibility; it will + # move here when docval is removed + from ..utils import get_docval_macro + try: + return tuple(t for t in get_docval_macro(macro_name) if isinstance(t, type)) + except KeyError: + return () + + +def type_name_validator(name): + """A validator accepting values with ``name`` anywhere in their class MRO.""" + if name in _type_name_validators: + return _type_name_validators[name] + + def checker(value): + return matches_type_name(value, name) + + checker.__name__ = checker.__qualname__ = "is_" + "".join(c if c.isalnum() else "_" for c in name) + validator = _remember(Is[checker], {'docval_name': name}) + _type_name_validators[name] = validator + return validator + + +def shape_validator(shape): + """A validator checking array shape when the value's shape is determinable. + + Values whose shape cannot be read pass here; ``@validated`` applies the stricter + check (including the unwrap-by-argument-name fallback, which needs the argument + name and therefore cannot live in a type validator). + """ + if shape in _shape_validators: + return _shape_validators[shape] + + def checker(value): + from ..utils import get_data_shape + from ._shapes import _shape_okay_multi + valshape = get_data_shape(value) + if valshape is None: + return True + return _shape_okay_multi(valshape, shape) + + dims = "x".join("any" if d in (None,) else str(d) for d in shape) if all( + not isinstance(d, (tuple, list)) for d in shape) else "multi" + checker.__name__ = checker.__qualname__ = f"has_shape_{dims}" + validator = _remember(Is[checker], {'shape': shape}) + _shape_validators[shape] = validator + return validator diff --git a/src/hdmf/typing/migrate.py b/src/hdmf/typing/migrate.py new file mode 100644 index 000000000..57880a2e0 --- /dev/null +++ b/src/hdmf/typing/migrate.py @@ -0,0 +1,459 @@ +"""AST-based migration tool for converting ``@docval`` functions to type hints. + +Usage:: + + python -m hdmf.typing.migrate path/to/file.py # print migrated source + python -m hdmf.typing.migrate path/to/file.py --diff # show unified diff + python -m hdmf.typing.migrate path/to/file.py --in-place # rewrite the file + +The tool rewrites each ``@docval``-decorated function whose argument specs are +literal dicts: the decorator becomes ``@validated``, the ``**kwargs`` signature +becomes a real type-hinted signature, a Google-style docstring is generated from the +docval ``doc`` strings, and mechanical ``getargs``/``popargs`` body lines are +removed. Anything it cannot convert safely is marked with a ``# TODO(migrate):`` +comment for human review: + +- ``@docval(*get_docval(...))`` splice decorators (composition sites) +- mutable default values (converted to ``= None``; the body needs a None-guard) +- bodies that still reference ``kwargs`` after conversion (e.g. + ``super().__init__(**kwargs)``) +""" + +import argparse +import ast +import difflib +import sys +from dataclasses import dataclass, field +from pathlib import Path + +# docval type strings -> hdmf.typing alias names +MACRO_MAP = { + 'array_data': 'ArrayData', + 'scalar_data': 'ScalarData', + 'data': 'AnyData', + 'int': 'Int', + 'uint': 'UInt', + 'float': 'Float', + 'bool': 'Bool', +} + +# bare python numeric types are widened by docval's check_type; the aliases preserve that +NAME_MAP = { + 'int': 'Int', + 'float': 'Float', + 'bool': 'Bool', +} + +MAX_SIGNATURE_WIDTH = 115 + + +@dataclass +class MigratedArg: + name: str + hint: str + doc: str = '' + default_src: str | None = None # source text of the default, None if required + todos: list[str] = field(default_factory=list) + + +@dataclass +class MigratedFunction: + node: ast.FunctionDef + decorator: ast.Call + args: list[MigratedArg] + options: dict[str, str] # remaining docval options (source text), e.g. allow_positional + returns_doc: str | None + rtype_hint: str | None + allow_extra: bool + todos: list[str] = field(default_factory=list) + + +class _AliasTracker: + """Track which hdmf.typing names the migrated code needs to import.""" + + def __init__(self): + self.names = set() + + def use(self, name): + self.names.add(name) + return name + + +class DocvalMigrator: + """Migrates ``@docval``-decorated functions in a source file to ``@validated``.""" + + def __init__(self): + self.aliases = _AliasTracker() + self.typing_names = set() # names needed from the stdlib typing module + + # ---------------------------------------------------------------- parsing + + def _find_docval_functions(self, tree): + found = [] + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + for dec in node.decorator_list: + if (isinstance(dec, ast.Call) and isinstance(dec.func, ast.Name) + and dec.func.id == 'docval'): + found.append((node, dec)) + break + return found + + def _convert_type(self, node, arg): + """Convert a docval type expression AST node to a type hint source string.""" + if isinstance(node, ast.Constant) and isinstance(node.value, str): + if node.value in MACRO_MAP: + return self.aliases.use(MACRO_MAP[node.value]) + self.aliases.use('TypeName') + return f"TypeName[{node.value!r}]" + if isinstance(node, ast.Constant) and node.value is None: + # docval type None means "any type" + self.typing_names.add('Any') + return 'Any' + if isinstance(node, ast.Name): + if node.id in NAME_MAP: + return self.aliases.use(NAME_MAP[node.id]) + return node.id + if isinstance(node, ast.Attribute): + return ast.unparse(node) # e.g. np.ndarray + if isinstance(node, (ast.Tuple, ast.List)): + members = [self._convert_type(elt, arg) for elt in node.elts] + # dedupe (e.g. ('array_data', list) -> ArrayData already covers list at runtime, + # but keep both: the union is what the author wrote) + seen = [] + for m in members: + if m not in seen: + seen.append(m) + return ' | '.join(seen) + arg.todos.append(f"could not convert type expression: {ast.unparse(node)}") + return ast.unparse(node) + + def _convert_arg(self, spec_node): + """Convert one literal docval spec dict AST node to a MigratedArg, or None. + + Handles both ``{'name': ...}`` dict literals and ``dict(name=...)`` calls. + """ + keys = {} + if isinstance(spec_node, ast.Dict): + for k, v in zip(spec_node.keys, spec_node.values): + if not (isinstance(k, ast.Constant) and isinstance(k.value, str)): + return None + keys[k.value] = v + elif (isinstance(spec_node, ast.Call) and isinstance(spec_node.func, ast.Name) + and spec_node.func.id == 'dict' and not spec_node.args): + for kw in spec_node.keywords: + if kw.arg is None: # dict(**something) + return None + keys[kw.arg] = kw.value + else: + return None + if 'name' not in keys or not isinstance(keys['name'], ast.Constant): + return None + arg = MigratedArg(name=keys['name'].value, hint='') + + if 'doc' in keys: + doc_node = keys['doc'] + if isinstance(doc_node, ast.Constant) and isinstance(doc_node.value, str): + arg.doc = ' '.join(doc_node.value.split()) + else: + arg.doc = '' + arg.todos.append(f"non-literal doc: {ast.unparse(doc_node)}") + + if 'enum' in keys: + enum_src = ast.unparse(keys['enum']).strip('()[]') + self.typing_names.add('Literal') + arg.hint = f"Literal[{enum_src}]" + else: + arg.hint = self._convert_type(keys.get('type', ast.Constant(value=None)), arg) + + if 'shape' in keys: + self.aliases.use('Shaped') + arg.hint = f"Shaped[{arg.hint}, {ast.unparse(keys['shape'])}]" + + if 'default' in keys: + default_node = keys['default'] + default_src = ast.unparse(default_node) + is_mutable = (isinstance(default_node, (ast.List, ast.Dict, ast.Set)) + or (isinstance(default_node, ast.Call) + and isinstance(default_node.func, ast.Name) + and default_node.func.id in ('list', 'dict', 'set'))) + allow_none = ('allow_none' in keys + and isinstance(keys['allow_none'], ast.Constant) + and keys['allow_none'].value) + default_is_none = isinstance(default_node, ast.Constant) and default_node.value is None + if is_mutable: + arg.todos.append(f"default was {default_src}; docval deepcopied it per call — " + "add a None-guard in the body") + default_src = 'None' + default_is_none = True + if (default_is_none or allow_none) and 'None' not in arg.hint.split(' | ') and arg.hint != 'Any': + arg.hint = f"{arg.hint} | None" + arg.default_src = default_src + return arg + + def _convert_function(self, node, dec): + """Convert one @docval function. Returns a MigratedFunction, or None to skip.""" + func = MigratedFunction(node=node, decorator=dec, args=[], options={}, + returns_doc=None, rtype_hint=None, allow_extra=False) + for spec_node in dec.args: + if isinstance(spec_node, ast.Starred): + func.todos.append( + f"decorator splices other functions' specs ({ast.unparse(spec_node)}); " + "convert by hand") + return None + arg = self._convert_arg(spec_node) + if arg is None: + func.todos.append(f"non-literal argument spec: {ast.unparse(spec_node)}") + return None + func.args.append(arg) + + for kw in dec.keywords: + if kw.arg == 'returns': + if isinstance(kw.value, ast.Constant): + func.returns_doc = kw.value.value + else: + func.todos.append(f"non-literal returns doc dropped: {ast.unparse(kw.value)}; " + "add a Returns: docstring section by hand") + elif kw.arg == 'rtype': + if isinstance(kw.value, ast.Constant) and isinstance(kw.value.value, str): + func.rtype_hint = f"{kw.value.value!r}" + else: + func.rtype_hint = ast.unparse(kw.value) + elif kw.arg == 'is_method': + pass # real signatures make this obsolete + elif kw.arg == 'allow_extra': + if isinstance(kw.value, ast.Constant) and kw.value.value: + func.allow_extra = True + elif kw.arg in ('enforce_type', 'enforce_shape', 'allow_positional'): + func.options[kw.arg] = ast.unparse(kw.value) + elif kw.arg in ('func_name', 'doc'): + func.todos.append(f"docval option {kw.arg}={ast.unparse(kw.value)} has no " + "@validated equivalent; convert by hand") + return None + return func + + # --------------------------------------------------------------- emitting + + def _existing_description(self, node): + doc = ast.get_docstring(node) + return doc.strip() if doc else '' + + def _build_docstring(self, func, indent): + lines = [] + description = self._existing_description(func.node) + body_indent = indent + ' ' + lines.append(f'{body_indent}"""{description.splitlines()[0] if description else func.node.name}') + for extra in (description.splitlines()[1:] if description else []): + lines.append(f'{body_indent}{extra}' if extra.strip() else '') + if func.args: + lines.append('') + lines.append(f'{body_indent}Args:') + for arg in func.args: + lines.append(f'{body_indent} {arg.name}: {arg.doc}') + if func.returns_doc: + lines.append('') + lines.append(f'{body_indent}Returns:') + lines.append(f'{body_indent} {func.returns_doc}') + lines.append(f'{body_indent}"""') + return lines + + def _build_signature(self, func, indent): + node = func.node + params = [] + existing = [a.arg for a in node.args.args] + if existing and existing[0] in ('self', 'cls'): + params.append(existing[0]) + required = [a for a in func.args if a.default_src is None] + optional = [a for a in func.args if a.default_src is not None] + for arg in required: + params.append(f"{arg.name}: {arg.hint}") + for arg in optional: + params.append(f"{arg.name}: {arg.hint} = {arg.default_src}") + if func.allow_extra: + params.append('**kwargs') + rtype = func.rtype_hint + if node.returns is not None: # an existing return annotation wins over docval rtype + rtype = ast.unparse(node.returns) + ret = f" -> {rtype}" if rtype and node.name != '__init__' else '' + one_line = f"{indent}def {node.name}({', '.join(params)}){ret}:" + if len(one_line) <= MAX_SIGNATURE_WIDTH: + return [one_line] + arg_indent = ' ' * (len(indent) + len(f"def {node.name}(") ) + lines = [f"{indent}def {node.name}({params[0]},"] + for p in params[1:-1]: + lines.append(f"{arg_indent}{p},") + lines.append(f"{arg_indent}{params[-1]}){ret}:") + return lines + + def _build_decorator(self, func, indent): + if func.options: + opts = ', '.join(f"{k}={v}" for k, v in sorted(func.options.items())) + return f"{indent}@validated({opts})" + return f"{indent}@validated" + + def _rewrite_body(self, func, body_lines, indent): + """Rewrite mechanical getargs/popargs/kwargs[...] lines; flag remaining kwargs uses.""" + import re + argnames = {a.name for a in func.args} + getargs_pattern = re.compile( + r"^(?P\s*)\(?\s*(?P[\w.]+(?:\s*,\s*[\w.]+)*)\s*,?\s*\)?\s*=" + r"\s*(?:getargs|popargs)\(\s*" + r"(?P(?:'[^']+'|\"[^\"]+\")(?:\s*,\s*(?:'[^']+'|\"[^\"]+\"))*)\s*,\s*kwargs\s*,?\s*\)\s*$") + subscript_pattern = re.compile( + r"^(?P\s*)(?P[\w.]+)\s*=\s*kwargs\[\s*(?:'(?P\w+)'|\"(?P\w+)\")\s*\]\s*$") + + def rewrite(logical, lead): + """Rewrite one whitespace-squashed logical line, or return None to keep it.""" + m = getargs_pattern.match(logical) + if m: + targets = [t.strip() for t in m.group('targets').split(',')] + names = [n.strip().strip('\'"') for n in m.group('names').split(',')] + if len(targets) == len(names) and all(n in argnames for n in names): + if targets == names: + return [] # parameters are now real names; the line is redundant + return [f"{lead}{', '.join(targets)} = {', '.join(names)}"] + m = subscript_pattern.match(logical) + if m: + name = m.group('name') or m.group('name2') + if name in argnames: + if m.group('target') == name: + return [] + return [f"{lead}{m.group('target')} = {name}"] + return None + + out = [] + buffer = [] + balance = 0 + for line in body_lines: + buffer.append(line) + balance += sum(line.count(c) for c in '([{') - sum(line.count(c) for c in ')]}') + if balance > 0 or line.rstrip().endswith('\\'): + continue # statement continues on the next line + balance = 0 + lead = re.match(r'\s*', buffer[0]).group() + logical = ' '.join(part.strip().rstrip('\\').strip() for part in buffer) + replacement = rewrite(lead + logical, lead) + out.extend(buffer if replacement is None else replacement) + buffer = [] + out.extend(buffer) # unbalanced trailing lines, if any + leftover_kwargs = any('kwargs' in line for line in out) and not func.allow_extra + todos = list(func.todos) + for arg in func.args: + todos.extend(f"{arg.name}: {t}" for t in arg.todos) + if leftover_kwargs: + todos.append("body still references `kwargs`, which no longer exists; rewrite " + "remaining uses (e.g. multi-line getargs/popargs, " + "super().__init__(**kwargs) -> explicit keywords)") + todo_lines = [f"{indent} # TODO(migrate): {t}" for t in todos] + return todo_lines + out + + # ------------------------------------------------------------------ driver + + def migrate_source(self, source): + """Return (migrated_source, n_converted, n_skipped).""" + tree = ast.parse(source) + targets = self._find_docval_functions(tree) + if not targets: + return source, 0, 0 + lines = source.splitlines() + n_converted = 0 + n_skipped = 0 + # bottom-up so earlier line numbers stay valid + for node, dec in sorted(targets, key=lambda t: t[0].lineno, reverse=True): + func = self._convert_function(node, dec) + if func is None: + n_skipped += 1 + dec_line = dec.lineno - 1 + indent = ' ' * (len(lines[dec_line]) - len(lines[dec_line].lstrip())) + lines.insert(dec_line, f"{indent}# TODO(migrate): docval decorator could not be " + "converted automatically; convert by hand") + continue + n_converted += 1 + # replace ALL decorator lines: every decorator is re-emitted in new_block, so + # starting any lower would leave duplicates (e.g. a doubled @classmethod, which + # Python 3.13+ rejects at call time) + dec_start = min(d.lineno for d in node.decorator_list) - 1 + # skip an existing docstring; it is folded into the generated one + body_start = node.body[0].lineno - 1 + body_end = node.end_lineno - 1 + first_stmt = node.body[0] + if (isinstance(first_stmt, ast.Expr) and isinstance(first_stmt.value, ast.Constant) + and isinstance(first_stmt.value.value, str)): + body_start = first_stmt.end_lineno # first line after the docstring + indent = ' ' * node.col_offset + + other_decorators = [ast.unparse(d) for d in node.decorator_list if d is not dec] + new_block = [f"{indent}@{d}" for d in other_decorators] + new_block.append(self._build_decorator(func, indent)) + new_block.extend(self._build_signature(func, indent)) + new_block.extend(self._build_docstring(func, indent)) + body_lines = lines[body_start:body_end + 1] if body_start <= body_end else [] + new_block.extend(self._rewrite_body(func, body_lines, indent)) + + lines[dec_start:body_end + 1] = new_block + migrated = '\n'.join(lines) + if source.endswith('\n') and not migrated.endswith('\n'): + migrated += '\n' + migrated = self._add_imports(migrated) + return migrated, n_converted, n_skipped + + def _add_imports(self, source): + stmts = [] + if self.typing_names and not any( + 'from typing import' in line and name in line + for name in self.typing_names for line in source.splitlines()): + stmts.append(f"from typing import {', '.join(sorted(self.typing_names))}") + if 'from hdmf.typing import' not in source: + import_names = sorted(self.aliases.names) + ['validated'] + stmts.append(f"from hdmf.typing import {', '.join(import_names)}") + if not stmts: + return source + lines = source.splitlines() + insert_at = 0 + for i, line in enumerate(lines): + if line.startswith(('import ', 'from ')): + insert_at = i + 1 + lines[insert_at:insert_at] = stmts + out = '\n'.join(lines) + return out + '\n' if source.endswith('\n') else out + + def migrate_file(self, filepath, in_place=False): + filepath = Path(filepath) + source = filepath.read_text() + migrated, n_converted, n_skipped = self.migrate_source(source) + if in_place and migrated != source: + filepath.write_text(migrated) + return migrated, n_converted, n_skipped + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog='python -m hdmf.typing.migrate', + description="Migrate @docval decorators to @validated with type hints") + parser.add_argument('files', nargs='+', help="Python files to migrate") + parser.add_argument('--diff', action='store_true', help="show a unified diff of the changes") + parser.add_argument('--in-place', '-i', action='store_true', help="rewrite files in place") + args = parser.parse_args(argv) + + migrator = DocvalMigrator() + for filepath in args.files: + filepath = Path(filepath) + if not filepath.exists(): + print(f"error: file not found: {filepath}", file=sys.stderr) # noqa: T201 + return 1 + original = filepath.read_text() + migrated, n_converted, n_skipped = migrator.migrate_file(filepath, in_place=args.in_place) + print(f"{filepath}: converted {n_converted}, needs manual attention {n_skipped}", # noqa: T201 + file=sys.stderr) + if args.diff: + sys.stdout.writelines(difflib.unified_diff( + original.splitlines(keepends=True), migrated.splitlines(keepends=True), + fromfile=f"a/{filepath.name}", tofile=f"b/{filepath.name}")) + elif not args.in_place: + print(migrated) # noqa: T201 + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/src/hdmf/typing/testing.py b/src/hdmf/typing/testing.py new file mode 100644 index 000000000..724589c47 --- /dev/null +++ b/src/hdmf/typing/testing.py @@ -0,0 +1,165 @@ +"""Helpers for verifying that a function migrated from ``@docval`` behaves identically. + +Use these in migration PRs (in hdmf and in downstream libraries): keep a copy of the +old ``@docval``-decorated function, migrate the real one, and assert parity over a +set of call cases. +""" + +import typing +from typing import Any, Callable + +from ..utils import get_docval + + +def compare_validation_behavior( + old_func: Callable, + new_func: Callable, + test_cases: list[dict[str, Any]], + *, + check_return_values: bool = True, + check_error_types: bool = True, +) -> list[dict[str, Any]]: + """Run both functions over the test cases and report any behavioral differences. + + Args: + old_func: the original ``@docval``-decorated function + new_func: the migrated type-hinted function + test_cases: list of dicts with optional 'args' (tuple) and 'kwargs' (dict) keys + check_return_values: also compare return values on success + check_error_types: also compare exception types on failure + + Returns: + One result dict per case; ``result['match']`` is False where behavior diverged. + """ + results = [] + for i, case in enumerate(test_cases): + args = case.get('args', ()) + kwargs = case.get('kwargs', {}) + result = {'case_index': i, 'args': args, 'kwargs': kwargs, + 'old_result': None, 'new_result': None, + 'old_error': None, 'new_error': None, + 'match': True, 'details': []} + try: + result['old_result'] = old_func(*args, **kwargs) + except Exception as e: + result['old_error'] = (type(e), str(e)) + try: + result['new_result'] = new_func(*args, **kwargs) + except Exception as e: + result['new_error'] = (type(e), str(e)) + + if result['old_error'] is None and result['new_error'] is None: + if check_return_values and result['old_result'] != result['new_result']: + result['match'] = False + result['details'].append( + f"return values differ: {result['old_result']!r} != {result['new_result']!r}") + elif result['old_error'] is not None and result['new_error'] is not None: + if check_error_types and result['old_error'][0] is not result['new_error'][0]: + result['match'] = False + result['details'].append( + f"error types differ: {result['old_error'][0].__name__} != " + f"{result['new_error'][0].__name__}") + else: + result['match'] = False + if result['old_error'] is None: + result['details'].append(f"old succeeded but new failed with: {result['new_error']}") + else: + result['details'].append(f"new succeeded but old failed with: {result['old_error']}") + results.append(result) + return results + + +def assert_validation_parity(old_func, new_func, test_cases, **kwargs): + """Assert that both functions behave identically over the test cases.""" + results = compare_validation_behavior(old_func, new_func, test_cases, **kwargs) + mismatches = [r for r in results if not r['match']] + if mismatches: + lines = [] + for r in mismatches: + lines.append(f"case {r['case_index']} (args={r['args']!r}, kwargs={r['kwargs']!r}): " + + '; '.join(r['details'])) + raise AssertionError("validation behavior diverged:\n" + '\n'.join(lines)) + + +def compare_docval_specs(old_func: Callable, new_func: Callable) -> dict[str, Any]: + """Compare ``get_docval`` output between two functions. + + Works for any mix of ``@docval``-decorated and type-hinted functions, since + :func:`hdmf.utils.get_docval` serves both. Verifies argument names, types, + defaults, shapes, enums, and docs. + """ + result = {'match': True, 'differences': []} + old_by_name = {a['name']: a for a in get_docval(old_func)} + new_by_name = {a['name']: a for a in get_docval(new_func)} + + missing = set(old_by_name) - set(new_by_name) + extra = set(new_by_name) - set(old_by_name) + if missing: + result['match'] = False + result['differences'].append(f"missing in new: {sorted(missing)}") + if extra: + result['match'] = False + result['differences'].append(f"extra in new: {sorted(extra)}") + + for name in set(old_by_name) & set(new_by_name): + old_arg, new_arg = old_by_name[name], new_by_name[name] + for key in ('type', 'default', 'shape', 'enum', 'doc'): + old_val = old_arg.get(key, '') + new_val = new_arg.get(key, '') + if key == 'type': + old_val = _normalize_type(old_val) + new_val = _normalize_type(new_val) + if old_val != new_val: + result['match'] = False + result['differences'].append( + f"arg '{name}' {key} differs: {old_val!r} != {new_val!r}") + return result + + +def _normalize_type(argtype): + """Normalize a docval type expression for comparison. + + docval resolves macro strings to type tuples at decoration time while the + synthesizer intentionally keeps the macro string; normalize both to a comparable + frozenset of names. + """ + from ..utils import get_docval_macro + if argtype is None: + return None + if isinstance(argtype, str): + try: + expanded = get_docval_macro(argtype) + except KeyError: + return frozenset({argtype}) + return frozenset(t.__name__ for t in expanded) + if isinstance(argtype, type): + return frozenset({argtype.__name__}) + if isinstance(argtype, (list, tuple)): + out = set() + for t in argtype: + out.update(_normalize_type(t)) + return frozenset(out) + return frozenset({str(argtype)}) + + +def find_required_nullable_params(func: Callable) -> list[str]: + """Lint helper: return names of required parameters hinted ``T | None``. + + docval cannot express a required-but-nullable argument, so these hints are lossy + for downstream splicing; give the parameter a default or drop ``| None``. + """ + import inspect + import types as _types + + from ._compat import _NoneType, _safe_hints + sig = inspect.signature(func) + hints = _safe_hints(func) + flagged = [] + for name, param in sig.parameters.items(): + if name in ('self', 'cls') or param.default is not inspect.Parameter.empty: + continue + hint = hints.get(name) + if typing.get_origin(hint) in (typing.Union, _types.UnionType) \ + and _NoneType in typing.get_args(hint): + flagged.append(name) + return flagged diff --git a/src/hdmf/utils.py b/src/hdmf/utils.py index 63dbb7e93..f5cd0042a 100644 --- a/src/hdmf/utils.py +++ b/src/hdmf/utils.py @@ -460,6 +460,10 @@ def __parse_args(validator, args, kwargs, enforce_type=True, enforce_shape=True, def get_docval(func, *args): '''Get a copy of docval arguments for a function. If args are supplied, return only docval arguments with value for 'name' key equal to the args + + For functions that are not decorated with ``@docval`` but have type-hinted + signatures, equivalent argument specs are synthesized from the signature, type + hints, and Google-style docstring (see :mod:`hdmf.typing`). ''' func_docval = getattr(func, docval_attr_name, None) if func_docval: @@ -470,12 +474,32 @@ def get_docval(func, *args): except KeyError as ke: raise ValueError('Function %s does not have docval argument %s' % (func.__name__, str(ke))) return tuple(func_docval[__docval_args_loc]) + elif __has_annotated_signature(func): + from .typing._compat import synthesize_docval # lazy import to avoid circularity + specs, idx, _ = synthesize_docval(func) + if args: + try: + return tuple(idx[name] for name in args) + except KeyError as ke: + raise ValueError('Function %s does not have docval argument %s' % (func.__name__, str(ke))) + return specs else: if args: raise ValueError('Function %s has no docval arguments' % func.__name__) return tuple() +def __has_annotated_signature(func): + """Return True if ``func`` is a plain function/method with type-hinted parameters.""" + target = getattr(func, '__func__', func) + while hasattr(target, '__wrapped__'): + target = target.__wrapped__ + annotations = getattr(target, '__annotations__', None) + if not annotations: + return False + return any(name != 'return' for name in annotations) + + def __resolve_type(t): if t is None: return t diff --git a/src/hdmf/validate/validator.py b/src/hdmf/validate/validator.py index c7709e80b..45cf304f0 100644 --- a/src/hdmf/validate/validator.py +++ b/src/hdmf/validate/validator.py @@ -14,8 +14,9 @@ from ..spec import SpecNamespace from ..spec.spec import BaseStorageSpec, DtypeHelper from ..utils import _is_collection, _get_length -from ..utils import docval, getargs, pystr, get_data_shape +from ..utils import pystr, get_data_shape from ..query import ReferenceResolver +from ..typing import validated __allowable = DtypeHelper.allowable @@ -252,9 +253,14 @@ def check_shape(expected, received): class ValidatorMap: """A class for keeping track of Validator objects for all data types in a namespace""" - @docval({'name': 'namespace', 'type': SpecNamespace, 'doc': 'the namespace to builder map for'}) - def __init__(self, **kwargs): - ns = getargs('namespace', kwargs) + @validated + def __init__(self, namespace: SpecNamespace): + """Initialize the ValidatorMap. + + Args: + namespace: the namespace to builder map for + """ + ns = namespace self.__ns = ns tree = defaultdict(list) types = ns.get_registered_types() @@ -295,11 +301,16 @@ def __rec(self, tree, node): def namespace(self): return self.__ns - @docval({'name': 'spec', 'type': (Spec, str), 'doc': 'the specification to use to validate'}, - returns='all valid sub data types for the given spec', rtype=tuple) - def valid_types(self, **kwargs): - '''Get all valid types for a given data type''' - spec = getargs('spec', kwargs) + @validated + def valid_types(self, spec: Spec | str) -> tuple: + """Get all valid types for a given data type + + Args: + spec: the specification to use to validate + + Returns: + all valid sub data types for the given spec + """ if isinstance(spec, Spec): spec = spec.data_type_def try: @@ -307,12 +318,17 @@ def valid_types(self, **kwargs): except KeyError: raise ValueError("no children for '%s'" % spec) - @docval({'name': 'data_type', 'type': (BaseStorageSpec, str), - 'doc': 'the data type to get the validator for'}, - returns='the validator ``data_type``') - def get_validator(self, **kwargs): - """Return the validator for a given data type""" - dt = getargs('data_type', kwargs) + @validated + def get_validator(self, data_type: BaseStorageSpec | str): + """Return the validator for a given data type + + Args: + data_type: the data type to get the validator for + + Returns: + the validator ``data_type`` + """ + dt = data_type if isinstance(dt, BaseStorageSpec): dt_tmp = dt.data_type_def if dt_tmp is None: @@ -324,15 +340,19 @@ def get_validator(self, **kwargs): msg = "data type '%s' not found in namespace %s" % (dt, self.__ns.name) raise ValueError(msg) - @docval({'name': 'builder', 'type': BaseBuilder, 'doc': 'the builder to validate'}, - returns="a list of errors found", rtype=list) - def validate(self, **kwargs): + @validated + def validate(self, builder: BaseBuilder) -> list: """Validate a builder against a Spec ``builder`` must have the attribute used to specifying data type by the namespace used to construct this ValidatorMap. + + Args: + builder: the builder to validate + + Returns: + a list of errors found """ - builder = getargs('builder', kwargs) dt = builder.attributes.get(self.__type_key) if dt is None: msg = "builder must have data type defined with attribute '%s'" % self.__type_key @@ -344,11 +364,16 @@ def validate(self, **kwargs): class Validator(metaclass=ABCMeta): '''A base class for classes that will be used to validate against Spec subclasses''' - @docval({'name': 'spec', 'type': Spec, 'doc': 'the specification to use to validate'}, - {'name': 'validator_map', 'type': ValidatorMap, 'doc': 'the ValidatorMap to use during validation'}) - def __init__(self, **kwargs): - self.__spec = getargs('spec', kwargs) - self.__vmap = getargs('validator_map', kwargs) + @validated + def __init__(self, spec: Spec, validator_map: ValidatorMap): + """Initialize the validator. + + Args: + spec: the specification to use to validate + validator_map: the ValidatorMap to use during validation + """ + self.__spec = spec + self.__vmap = validator_map @property def spec(self): @@ -359,9 +384,16 @@ def vmap(self): return self.__vmap @abstractmethod - @docval({'name': 'value', 'type': None, 'doc': 'either in the form of a value or a Builder'}, - returns='a list of Errors', rtype=list) - def validate(self, **kwargs): + @validated + def validate(self, value: Any) -> list: + """Validate the given value against this validator's spec. + + Args: + value: either in the form of a value or a Builder + + Returns: + a list of Errors + """ pass @classmethod @@ -381,15 +413,26 @@ def get_builder_loc(cls, builder): class AttributeValidator(Validator): '''A class for validating values against AttributeSpecs''' - @docval({'name': 'spec', 'type': AttributeSpec, 'doc': 'the specification to use to validate'}, - {'name': 'validator_map', 'type': ValidatorMap, 'doc': 'the ValidatorMap to use during validation'}) - def __init__(self, **kwargs): - super().__init__(**kwargs) + @validated + def __init__(self, spec: AttributeSpec, validator_map: ValidatorMap): + """Initialize the validator. + + Args: + spec: the specification to use to validate + validator_map: the ValidatorMap to use during validation + """ + super().__init__(spec=spec, validator_map=validator_map) + + @validated + def validate(self, value: Any) -> list: + """Validate the given value against this validator's spec. - @docval({'name': 'value', 'type': None, 'doc': 'the value to validate'}, - returns='a list of Errors', rtype=list) - def validate(self, **kwargs): - value = getargs('value', kwargs) + Args: + value: the value to validate + + Returns: + a list of Errors + """ ret = list() spec = self.spec @@ -445,18 +488,29 @@ def validate(self, **kwargs): class BaseStorageValidator(Validator): '''A base class for validating against Spec objects that have attributes i.e. BaseStorageSpec''' - @docval({'name': 'spec', 'type': BaseStorageSpec, 'doc': 'the specification to use to validate'}, - {'name': 'validator_map', 'type': ValidatorMap, 'doc': 'the ValidatorMap to use during validation'}) - def __init__(self, **kwargs): - super().__init__(**kwargs) + @validated + def __init__(self, spec: BaseStorageSpec, validator_map: ValidatorMap): + """Initialize the validator. + + Args: + spec: the specification to use to validate + validator_map: the ValidatorMap to use during validation + """ + super().__init__(spec=spec, validator_map=validator_map) self.__attribute_validators = dict() for attr in self.spec.attributes: self.__attribute_validators[attr.name] = AttributeValidator(attr, self.vmap) - @docval({"name": "builder", "type": BaseBuilder, "doc": "the builder to validate"}, - returns='a list of Errors', rtype=list) - def validate(self, **kwargs): - builder = getargs('builder', kwargs) + @validated + def validate(self, builder: BaseBuilder) -> list: + """Validate the given value against this validator's spec. + + Args: + builder: the builder to validate + + Returns: + a list of Errors + """ attributes = builder.attributes ret = list() for attr, validator in self.__attribute_validators.items(): @@ -476,10 +530,15 @@ def validate(self, **kwargs): class DatasetValidator(BaseStorageValidator): '''A class for validating DatasetBuilders against DatasetSpecs''' - @docval({'name': 'spec', 'type': DatasetSpec, 'doc': 'the specification to use to validate'}, - {'name': 'validator_map', 'type': ValidatorMap, 'doc': 'the ValidatorMap to use during validation'}) - def __init__(self, **kwargs): - super().__init__(**kwargs) + @validated + def __init__(self, spec: DatasetSpec, validator_map: ValidatorMap): + """Initialize the validator. + + Args: + spec: the specification to use to validate + validator_map: the ValidatorMap to use during validation + """ + super().__init__(spec=spec, validator_map=validator_map) def _check_ref_target_type(self, val, expected_type, type_key, builder, ret): """Helper to recursively validate reference target types and hierarchy.""" @@ -502,10 +561,16 @@ def _check_ref_target_type(self, val, expected_type, type_key, builder, ret): for v in val: self._check_ref_target_type(v, expected_type, type_key, builder, ret) - @docval({"name": "builder", "type": DatasetBuilder, "doc": "the builder to validate"}, - returns='a list of Errors', rtype=list) - def validate(self, **kwargs): - builder = getargs('builder', kwargs) + @validated + def validate(self, builder: DatasetBuilder) -> list: + """Validate the given value against this validator's spec. + + Args: + builder: the builder to validate + + Returns: + a list of Errors + """ ret = super().validate(builder) data = builder.data if self.spec.dtype is not None: @@ -575,15 +640,26 @@ def _resolve_data_type(spec): class GroupValidator(BaseStorageValidator): '''A class for validating GroupBuilders against GroupSpecs''' - @docval({'name': 'spec', 'type': GroupSpec, 'doc': 'the specification to use to validate'}, - {'name': 'validator_map', 'type': ValidatorMap, 'doc': 'the ValidatorMap to use during validation'}) - def __init__(self, **kwargs): - super().__init__(**kwargs) + @validated + def __init__(self, spec: GroupSpec, validator_map: ValidatorMap): + """Initialize the validator. + + Args: + spec: the specification to use to validate + validator_map: the ValidatorMap to use during validation + """ + super().__init__(spec=spec, validator_map=validator_map) + + @validated + def validate(self, builder: GroupBuilder) -> list: + """Validate the given value against this validator's spec. - @docval({"name": "builder", "type": GroupBuilder, "doc": "the builder to validate"}, - returns='a list of Errors', rtype=list) - def validate(self, **kwargs): - builder = getargs('builder', kwargs) + Args: + builder: the builder to validate + + Returns: + a list of Errors + """ errors = super().validate(builder) errors.extend(self.__validate_children(builder)) return self._remove_duplicates(errors) diff --git a/tests/unit/typing_tests/__init__.py b/tests/unit/typing_tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/typing_tests/test_migrate.py b/tests/unit/typing_tests/test_migrate.py new file mode 100644 index 000000000..56a0b3502 --- /dev/null +++ b/tests/unit/typing_tests/test_migrate.py @@ -0,0 +1,118 @@ +"""Tests for the @docval -> @validated migration tool.""" + +import textwrap + +from hdmf.testing import TestCase +from hdmf.typing.migrate import DocvalMigrator + + +class TestDocvalMigrator(TestCase): + + def _migrate(self, source): + migrator = DocvalMigrator() + return migrator.migrate_source(textwrap.dedent(source)) + + def test_basic_conversion(self): + migrated, n_converted, n_skipped = self._migrate('''\ + from hdmf.utils import docval, getargs + + + class Thing: + + @docval({'name': 'name', 'type': str, 'doc': 'the name'}, + {'name': 'count', 'type': 'int', 'doc': 'how many', 'default': 1}) + def __init__(self, **kwargs): + name, count = getargs('name', 'count', kwargs) + self.name = name + self.count = count + ''') + self.assertEqual(n_converted, 1) + self.assertEqual(n_skipped, 0) + self.assertIn('@validated', migrated) + self.assertIn('def __init__(self, name: str, count: Int = 1):', migrated) + self.assertIn('name: the name', migrated) + self.assertNotIn('getargs', migrated.split('import')[-1].split('\n', 1)[1]) + self.assertIn('from hdmf.typing import Int, validated', migrated) + # the migrated module must be valid python + compile(migrated, '', 'exec') + + def test_macro_shape_enum_conversion(self): + migrated, n_converted, _ = self._migrate('''\ + from hdmf.utils import docval, getargs + + @docval({'name': 'data', 'type': ('array_data', 'data'), 'doc': 'd', 'shape': (None, 3)}, + {'name': 'mode', 'type': str, 'doc': 'm', 'enum': ['r', 'w'], 'default': 'r'}, + is_method=False) + def func(**kwargs): + data, mode = getargs('data', 'mode', kwargs) + return data + ''') + self.assertEqual(n_converted, 1) + self.assertIn('data: Shaped[ArrayData | AnyData, (None, 3)]', migrated) + self.assertIn("mode: Literal['r', 'w'] = 'r'", migrated) + self.assertIn('from typing import Literal', migrated) + compile(migrated, '', 'exec') + + def test_mutable_default_flagged(self): + migrated, n_converted, _ = self._migrate('''\ + from hdmf.utils import docval, getargs + + @docval({'name': 'tags', 'type': list, 'doc': 't', 'default': list()}, is_method=False) + def func(**kwargs): + tags = getargs('tags', kwargs) + return tags + ''') + self.assertEqual(n_converted, 1) + self.assertIn('tags: list | None = None', migrated) + self.assertIn('TODO(migrate)', migrated) + self.assertIn('None-guard', migrated) + + def test_splice_decorator_skipped_with_todo(self): + migrated, n_converted, n_skipped = self._migrate('''\ + from hdmf.utils import docval, get_docval, getargs + + @docval({'name': 'a', 'type': str, 'doc': 'a'}, is_method=False) + def parent(**kwargs): + return getargs('a', kwargs) + + @docval(*get_docval(parent), {'name': 'b', 'type': str, 'doc': 'b'}, is_method=False) + def child(**kwargs): + return getargs('a', 'b', kwargs) + ''') + self.assertEqual(n_converted, 1) + self.assertEqual(n_skipped, 1) + self.assertIn('convert by hand', migrated) + + def test_other_decorators_not_duplicated(self): + """@classmethod etc. must be re-emitted exactly once (doubling breaks on Python 3.13+).""" + migrated, n_converted, _ = self._migrate('''\ + from hdmf.utils import docval, getargs + + class C: + + @classmethod + @docval({'name': 'path', 'type': str, 'doc': 'the path'}) + def from_path(cls, **kwargs): + path = getargs('path', kwargs) + return cls(path) + ''') + self.assertEqual(n_converted, 1) + self.assertEqual(migrated.count('@classmethod'), 1) + self.assertLess(migrated.index('@classmethod'), migrated.index('@validated')) + compile(migrated, '', 'exec') + + def test_leftover_kwargs_flagged(self): + migrated, n_converted, _ = self._migrate('''\ + from hdmf.utils import docval, popargs + + class Sub(Base): + + @docval({'name': 'a', 'type': str, 'doc': 'a'}, + {'name': 'b', 'type': str, 'doc': 'b'}) + def __init__(self, **kwargs): + a = popargs('a', kwargs) + super().__init__(**kwargs) + self.a = a + ''') + self.assertEqual(n_converted, 1) + self.assertIn('body still references `kwargs`', migrated) diff --git a/tests/unit/typing_tests/test_typing_compat.py b/tests/unit/typing_tests/test_typing_compat.py new file mode 100644 index 000000000..cd709bdb5 --- /dev/null +++ b/tests/unit/typing_tests/test_typing_compat.py @@ -0,0 +1,419 @@ +"""Round-trip tests for the get_docval compatibility shim. + +The compatibility guarantee under test: for a type-hinted function, splicing +``get_docval(hinted_func, ...)`` into a legacy ``@docval`` decorator (the pervasive +downstream pattern, e.g. in PyNWB) must validate inputs exactly like the equivalent +hand-written docval spec. +""" + +import typing + +import numpy as np + +from hdmf.testing import TestCase +from hdmf.typing import AnyData, ArrayData, Bool, Float, Int, ScalarData, Shaped, TypeName, UInt +from hdmf.utils import docval, get_docval, getargs + + +class TestSynthesizedSpecs(TestCase): + """Test the docval spec dicts synthesized from type-hinted signatures.""" + + def test_basic_types(self): + def func(a: str, b: dict, c: bytes): + """F. + + Args: + a: doc a + b: doc b + c: doc c + """ + + self.assertTupleEqual(get_docval(func), ( + {'name': 'a', 'doc': 'doc a', 'type': str}, + {'name': 'b', 'doc': 'doc b', 'type': dict}, + {'name': 'c', 'doc': 'doc c', 'type': bytes}, + )) + + def test_numeric_aliases(self): + def func(a: Int, b: UInt, c: Float, d: Bool): + """F. + + Args: + a: doc a + b: doc b + c: doc c + d: doc d + """ + + types = [spec['type'] for spec in get_docval(func)] + self.assertListEqual(types, ['int', 'uint', 'float', 'bool']) + + def test_macro_aliases_stay_strings(self): + """Macro aliases must map to macro *strings* so late registrations apply.""" + def func(a: ArrayData, b: ScalarData, c: AnyData): + """F. + + Args: + a: doc a + b: doc b + c: doc c + """ + + types = [spec['type'] for spec in get_docval(func)] + self.assertListEqual(types, ['array_data', 'scalar_data', 'data']) + + def test_union(self): + def func(a: slice | list | tuple): + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0]['type'], (slice, list, tuple)) + + def test_optional_with_none_default(self): + def func(a: str | None = None): + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0], {'name': 'a', 'doc': 'doc a', 'type': str, 'default': None}) + + def test_optional_with_non_none_default_sets_allow_none(self): + def func(a: int | None = 5): + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0], + {'name': 'a', 'doc': 'doc a', 'type': int, 'default': 5, 'allow_none': True}) + + def test_bare_int_keeps_numpy_widening(self): + """Bare int/float/bool hints map to the classes, which docval's check_type widens.""" + def func(a: int, b: float, c: bool): + """F. + + Args: + a: doc a + b: doc b + c: doc c + """ + + types = [spec['type'] for spec in get_docval(func)] + self.assertListEqual(types, [int, float, bool]) + # widening parity after splice: np.int32 must pass for a bare int hint + @docval(*get_docval(func), is_method=False) + def downstream(**kwargs): + return kwargs + + downstream(a=np.int32(1), b=np.float32(0.5), c=np.bool_(True)) + + def test_literal_maps_to_enum(self): + def func(a: typing.Literal['r', 'w'] = 'r'): + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0], + {'name': 'a', 'doc': 'doc a', 'type': str, 'enum': ('r', 'w'), 'default': 'r'}) + + def test_shaped(self): + def func(a: Shaped[ArrayData, (None, 3)]): + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0], + {'name': 'a', 'doc': 'doc a', 'type': 'array_data', 'shape': (None, 3)}) + + def test_type_name(self): + def func(a: TypeName['DynamicTable']): # noqa: F821 + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0]['type'], 'DynamicTable') + + def test_unresolvable_forward_ref_kept_as_string(self): + def func(a: "NotARealClassAnywhere"): # noqa: F821 + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0]['type'], 'NotARealClassAnywhere') + + def test_parametrized_generic_degrades_to_origin(self): + def func(a: dict[str, int]): + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0]['type'], dict) + + def test_unannotated_param_is_any_type(self): + def func(a: str, b=None): + """F. + + Args: + a: doc a + b: doc b + """ + + self.assertEqual(get_docval(func, 'b')[0], {'name': 'b', 'doc': 'doc b', 'type': None, 'default': None}) + + def test_missing_docstring_doc_is_empty_string(self): + def func(a: str): + pass + + self.assertEqual(get_docval(func, 'a')[0], {'name': 'a', 'doc': '', 'type': str}) + + def test_self_and_kwargs_skipped(self): + class C: + def method(self, a: str, **kwargs): + """M. + + Args: + a: doc a + """ + + self.assertEqual([s['name'] for s in get_docval(C.method)], ['a']) + + def test_named_arg_selection_and_missing_name_raises(self): + def func(a: str, b: int = 1): + """F. + + Args: + a: doc a + b: doc b + """ + + self.assertEqual(get_docval(func, 'b')[0]['name'], 'b') + with self.assertRaisesRegex(ValueError, "does not have docval argument"): + get_docval(func, 'nonexistent') + + def test_no_annotations_legacy_behavior(self): + def func(a, b=None): + pass + + self.assertTupleEqual(get_docval(func), tuple()) + with self.assertRaisesRegex(ValueError, "has no docval arguments"): + get_docval(func, 'a') + + def test_star_args_rejected(self): + def func(a: str, *args): + pass + + with self.assertRaisesRegex(TypeError, r"\*args"): + get_docval(func) + + def test_docval_decorated_takes_precedence(self): + """A @docval function with annotations elsewhere must use the legacy path.""" + @docval({'name': 'a', 'type': str, 'doc': 'doc a'}) + def func(self, **kwargs): + pass + + self.assertEqual(get_docval(func, 'a')[0]['doc'], 'doc a') + + +class TestSpliceRoundTrip(TestCase): + """Splice synthesized specs into legacy @docval and verify validation parity.""" + + @staticmethod + def _make_downstream(*specs): + @docval(*specs, is_method=False) + def downstream(**kwargs): + return kwargs + return downstream + + def test_str_and_macro_splice(self): + def parent(name: str, data: ArrayData): + """P. + + Args: + name: the name + data: the data + """ + + downstream = self._make_downstream(*get_docval(parent)) + self.assertEqual(downstream(name='n', data=[1, 2])['name'], 'n') + downstream(name='n', data=np.arange(3)) # ndarray accepted by macro + with self.assertRaisesRegex(TypeError, "incorrect type for 'name'"): + downstream(name=5, data=[1, 2]) + with self.assertRaisesRegex(TypeError, "incorrect type for 'data'"): + downstream(name='n', data=5) + + def test_numpy_widening_after_splice(self): + def parent(count: Int, frac: Float, flag: Bool): + """P. + + Args: + count: c + frac: f + flag: g + """ + + downstream = self._make_downstream(*get_docval(parent)) + result = downstream(count=np.int32(5), frac=np.float32(0.5), flag=np.bool_(True)) + self.assertEqual(result['count'], 5) + with self.assertRaisesRegex(TypeError, "incorrect type for 'count'"): + downstream(count=5.0, frac=0.5, flag=True) + + def test_shape_after_splice(self): + def parent(data: Shaped[ArrayData, (None, 3)]): + """P. + + Args: + data: the data + """ + + downstream = self._make_downstream(*get_docval(parent)) + downstream(data=[[1, 2, 3], [4, 5, 6]]) + with self.assertRaisesRegex(ValueError, "incorrect shape for data"): + downstream(data=[[1, 2], [3, 4]]) + + def test_enum_after_splice(self): + def parent(mode: typing.Literal['r', 'w'] = 'r'): + """P. + + Args: + mode: the mode + """ + + downstream = self._make_downstream(*get_docval(parent)) + downstream(mode='w') + with self.assertRaisesRegex(ValueError, "forbidden value for 'mode'"): + downstream(mode='x') + + def test_default_after_splice(self): + def parent(count: Int = 7): + """P. + + Args: + count: c + """ + + downstream = self._make_downstream(*get_docval(parent)) + self.assertEqual(downstream()['count'], 7) + + def test_mixed_splice_with_new_docval_args(self): + """The canonical pynwb pattern: parent args spliced next to new spec dicts.""" + def parent(name: str, data: AnyData | None = None): + """P. + + Args: + name: the name + data: the data + """ + + @docval(*get_docval(parent, 'name'), + {'name': 'extra', 'type': 'int', 'doc': 'an extra', 'default': 0}, + *get_docval(parent, 'data'), + is_method=False) + def downstream(**kwargs): + return getargs('name', 'extra', 'data', kwargs) + + self.assertEqual(downstream(name='n'), ['n', 0, None]) + with self.assertRaisesRegex(TypeError, "incorrect type for 'extra'"): + downstream(name='n', extra='x') + + def test_type_name_after_splice(self): + from hdmf.common import DynamicTable + + def parent(table: TypeName['DynamicTable']): # noqa: F821 + """P. + + Args: + table: a table + """ + + downstream = self._make_downstream(*get_docval(parent)) + downstream(table=DynamicTable(name='t', description='d')) + with self.assertRaisesRegex(TypeError, "incorrect type for 'table'"): + downstream(table=5) + + def test_macro_registration_before_splice_applies(self): + """Macro registrations made before splice decoration must be honored. + + Note: docval resolves macro strings to concrete type tuples at decoration + time, so (exactly as in pure-docval code) registrations made *after* a + downstream class is decorated do not apply to it. Synthesized specs keep + the macro string, so resolution happens at each splice's decoration. + """ + from hdmf.utils import docval_macro + + def parent(data: TypeName['SpliceTestData']): + """P. + + Args: + data: the data + """ + + downstream = self._make_downstream(*get_docval(parent)) + + class SpliceTestData: + pass + + downstream(data=SpliceTestData()) # MRO-name matching needs no registration + + # macro string case: register a type, then splice — the new type is accepted + @docval_macro('data') + class RegisteredBeforeSplice: + pass + + def parent2(data: AnyData): + """P. + + Args: + data: the data + """ + + downstream2 = self._make_downstream(*get_docval(parent2)) + downstream2(data=RegisteredBeforeSplice()) + + +class TestDocvalSpecComparison(TestCase): + """Verify synthesized specs match hand-written docval specs for equivalent functions.""" + + def test_parity_with_hand_written_docval(self): + from hdmf.typing.testing import compare_docval_specs + + @docval({'name': 'name', 'type': str, 'doc': 'the name'}, + {'name': 'data', 'type': ('array_data', 'data'), 'doc': 'the data', 'shape': (None, 3)}, + {'name': 'count', 'type': 'int', 'doc': 'how many', 'default': 1}) + def old(self, **kwargs): + pass + + def new(name: str, data: Shaped[ArrayData | AnyData, (None, 3)], count: Int = 1): + """N. + + Args: + name: the name + data: the data + count: how many + """ + + result = compare_docval_specs(old, new) + self.assertTrue(result['match'], result['differences']) + + def test_required_nullable_lint(self): + from hdmf.typing.testing import find_required_nullable_params + + def func(a: str | None, b: int, c: str | None = None): + pass + + self.assertEqual(find_required_nullable_params(func), ['a']) diff --git a/tests/unit/typing_tests/test_validated.py b/tests/unit/typing_tests/test_validated.py new file mode 100644 index 000000000..89ae0b38a --- /dev/null +++ b/tests/unit/typing_tests/test_validated.py @@ -0,0 +1,507 @@ +"""Tests for the @validated decorator: docval-parity runtime validation of type hints.""" + +import typing + +import numpy as np + +from hdmf.testing import TestCase +from hdmf.typing import ( + AllowPositional, + AnyData, + ArrayData, + Float, + Int, + Shaped, + TypeName, + set_type_checking, + validated, +) +from hdmf.utils import get_docval + + +class TestValidatedTypes(TestCase): + + def test_basic_type_error_message_matches_docval(self): + @validated + def func(a: str): + """F. + + Args: + a: doc a + """ + return a + + self.assertEqual(func('x'), 'x') + with self.assertRaisesWith(TypeError, + "TestValidatedTypes.test_basic_type_error_message_matches_docval." + ".func: incorrect type for 'a' (got 'int', expected 'str')"): + func(5) + + def test_numpy_widening(self): + @validated + def func(a: Int, b: Float): + """F. + + Args: + a: doc a + b: doc b + """ + return a + + self.assertEqual(func(np.int64(3), np.float16(0.5)), 3) + self.assertEqual(func(3, 0.5), 3) + with self.assertRaisesRegex(TypeError, "incorrect type for 'a'"): + func(3.0, 0.5) + + def test_union(self): + @validated + def func(a: slice | list): + """F. + + Args: + a: doc a + """ + return a + + func(slice(0, 5)) + func([1, 2]) + with self.assertRaisesRegex(TypeError, "expected 'slice or list'"): + func((1, 2)) + + def test_macro_alias_uses_live_registry(self): + from hdmf.utils import docval_macro + + @validated + def func(a: AnyData): + """F. + + Args: + a: doc a + """ + return a + + @docval_macro('data') + class RegisteredAfterDecoration: + pass + + func(RegisteredAfterDecoration()) + + def test_none_default_allows_none(self): + @validated + def func(a: str | None = None): + """F. + + Args: + a: doc a + """ + return a + + self.assertIsNone(func()) + self.assertIsNone(func(None)) + self.assertEqual(func('x'), 'x') + + def test_none_for_required_arg(self): + @validated + def func(a: str): + """F. + + Args: + a: doc a + """ + return a + + with self.assertRaisesRegex(TypeError, "None is not allowed for 'a' \\(expected 'str', not None\\)"): + func(None) + + def test_shape(self): + @validated + def func(data: Shaped[ArrayData, ((None,), (None, None))]): + """F. + + Args: + data: doc + """ + return data + + func([1, 2, 3]) + func([[1], [2]]) + with self.assertRaisesRegex(ValueError, "incorrect shape for data"): + func([[[1]]]) + + def test_shape_unwraps_by_argname(self): + """docval parity: if the value's shape is unreadable, getattr(value, argname) is checked.""" + class HasData: + def __init__(self, data): + self.data = data + + @validated + def func(data: Shaped[typing.Any, (None, 3)]): + """F. + + Args: + data: doc + """ + return data + + func(HasData([[1, 2, 3]])) + with self.assertRaisesRegex(ValueError, "incorrect shape for data"): + func(HasData([[1, 2]])) + + def test_enum(self): + @validated + def func(mode: typing.Literal['r', 'w', 'a']): + """F. + + Args: + mode: doc + """ + return mode + + self.assertEqual(func('w'), 'w') + with self.assertRaisesRegex(ValueError, "forbidden value for 'mode' \\(got 'x', expected"): + func('x') + + def test_type_name_mro_matching(self): + @validated + def func(thing: TypeName['MroTestTarget']): + """F. + + Args: + thing: doc + """ + return thing + + class MroTestTarget: + pass + + class Subclass(MroTestTarget): + pass + + func(MroTestTarget()) + func(Subclass()) + with self.assertRaisesRegex(TypeError, "expected 'MroTestTarget'"): + func(5) + + def test_multiple_errors_aggregated(self): + @validated + def func(a: str, b: Int): + """F. + + Args: + a: doc a + b: doc b + """ + + with self.assertRaisesRegex( + TypeError, + "incorrect type for 'a' \\(got 'int', expected 'str'\\), " + "incorrect type for 'b' \\(got 'str', expected 'int'\\)"): + func(5, 'x') + + def test_term_set_wrapper_passes_through(self): + from hdmf.term_set import TermSetWrapper + + # a minimal object satisfying TermSetWrapper's interface is complex to build; + # instead check that the wrapper type is unwrapped for validation using __new__ + wrapper = TermSetWrapper.__new__(TermSetWrapper) + wrapper.__dict__['_TermSetWrapper__value'] = 'wrapped-value' + + @validated + def func(a: str): + """F. + + Args: + a: doc a + """ + return a + + result = func(wrapper) + self.assertIs(result, wrapper) # wrapper object, not the unwrapped value, reaches the body + + +class TestValidatedCallConventions(TestCase): + + def test_positional_warning(self): + @validated(allow_positional=AllowPositional.WARNING) + def func(a: str): + """F. + + Args: + a: doc a + """ + return a + + with self.assertWarnsRegex(FutureWarning, "Using positional arguments for this method is discouraged"): + func('x') + + def test_positional_error(self): + @validated(allow_positional=AllowPositional.ERROR) + def func(a: str): + """F. + + Args: + a: doc a + """ + return a + + with self.assertRaisesRegex(SyntaxError, "Only keyword arguments"): + func('x') + self.assertEqual(func(a='x'), 'x') + + def test_method_self_not_counted_as_positional(self): + class C: + @validated(allow_positional=AllowPositional.ERROR) + def method(self, a: str = 'd'): + """M. + + Args: + a: doc a + """ + return a + + self.assertEqual(C().method(a='x'), 'x') # self alone must not trigger the error + + def test_missing_and_unrecognized_args(self): + @validated + def func(a: str): + """F. + + Args: + a: doc a + """ + return a + + with self.assertRaisesRegex(TypeError, "missing a required argument: 'a'"): + func() + with self.assertRaisesRegex(TypeError, "unexpected keyword argument 'b'"): + func(a='x', b=1) + + def test_kwargs_extras_pass_through_unvalidated(self): + @validated + def func(a: str, **kwargs): + """F. + + Args: + a: doc a + """ + return kwargs + + sentinel = object() + self.assertEqual(func(a='x', anything=sentinel), {'anything': sentinel}) + + def test_keyword_only_params(self): + @validated + def func(*, a: str): + """F. + + Args: + a: doc a + """ + return a + + self.assertEqual(func(a='x'), 'x') + with self.assertRaisesRegex(TypeError, "too many positional arguments"): + func('x') + + def test_enforce_flags(self): + @validated(enforce_type=False, enforce_shape=False) + def func(a: str, data: Shaped[ArrayData, (None, 3)] = None): + """F. + + Args: + a: doc a + data: doc + """ + return a + + self.assertEqual(func(5, data=[[1, 2]]), 5) + + def test_type_checking_kill_switch(self): + @validated + def func(a: str): + """F. + + Args: + a: doc a + """ + return a + + set_type_checking(False) + try: + self.assertEqual(func(5), 5) # no validation + finally: + set_type_checking(True) + with self.assertRaises(TypeError): + func(5) + + def test_wrapper_metadata(self): + @validated + def func(a: str): + """The description. + + Args: + a: doc a + """ + return a + + self.assertEqual(func.__name__, 'func') + self.assertIn('The description.', func.__doc__) + self.assertEqual(func.__validated__['args'][0]['name'], 'a') + + def test_get_docval_on_validated_function(self): + @validated + def func(a: str, data: ArrayData | None = None): + """F. + + Args: + a: doc a + data: the data + """ + + self.assertEqual(get_docval(func, 'a')[0], {'name': 'a', 'doc': 'doc a', 'type': str}) + self.assertEqual(get_docval(func, 'data')[0], + {'name': 'data', 'doc': 'the data', 'type': 'array_data', 'default': None}) + + +class TestValidatedLossyHints(TestCase): + """Hints docval cannot express are validated against the original hint via beartype.""" + + def test_parametrized_generic_element_check(self): + @validated + def func(m: dict[str, int]): + """F. + + Args: + m: doc m + """ + return m + + func({'a': 1}) + with self.assertRaisesRegex(TypeError, "incorrect type for 'm'"): + func({'a': 'not-an-int'}) + + def test_numpydantic_ndarray(self): + from numpydantic import NDArray, Shape + + @validated + def func(data: NDArray[Shape["* x, 3 y"], np.int64]): # noqa: F722 + """F. + + Args: + data: doc + """ + return data + + func(np.zeros((4, 3), dtype=np.int64)) + with self.assertRaisesRegex(TypeError, "incorrect type for 'data'"): + func(np.zeros((4, 2), dtype=np.int64)) + + # the synthesized docval spec still carries an equivalent shape + self.assertEqual(get_docval(func, 'data')[0]['shape'], (None, 3)) + + +class TestBeartypeNativeEnforcement(TestCase): + """The hdmf.typing aliases are real beartype validators: they are enforced by + plain beartype (no @validated involved), because validation is built on the + type-hint system itself, not on docval.""" + + def test_macro_alias_enforced_by_plain_beartype(self): + from beartype import beartype + from beartype.roar import BeartypeCallHintParamViolation + + @beartype + def func(a: ArrayData): + return a + + func([1, 2, 3]) + func(np.arange(3)) + with self.assertRaises(BeartypeCallHintParamViolation): + func(5) + + def test_type_name_enforced_by_plain_beartype(self): + from beartype import beartype + from beartype.roar import BeartypeCallHintParamViolation + + @beartype + def func(thing: TypeName['BeartypeNativeTarget']): + return thing + + class BeartypeNativeTarget: + pass + + func(BeartypeNativeTarget()) + with self.assertRaises(BeartypeCallHintParamViolation): + func(5) + + def test_shaped_enforced_by_plain_beartype(self): + from beartype import beartype + from beartype.roar import BeartypeCallHintParamViolation + + @beartype + def func(data: Shaped[ArrayData, (None, 3)]): + return data + + func([[1, 2, 3], [4, 5, 6]]) + with self.assertRaises(BeartypeCallHintParamViolation): + func([[1, 2], [3, 4]]) + + def test_bare_int_hint_is_strict(self): + """A bare `int` hint has standard type-hint semantics: numpy ints are + rejected. Use hdmf.typing.Int for numpy widening.""" + @validated + def func(a: int): + """F. + + Args: + a: doc a + """ + return a + + self.assertEqual(func(5), 5) + with self.assertRaisesRegex(TypeError, "incorrect type for 'a'"): + func(np.int32(5)) + + def test_int_alias_in_union_collapses_in_spec(self): + """Int | str synthesizes to docval ('int', str), not the raw numpy union.""" + def func(a: Int | str): + """F. + + Args: + a: doc a + """ + + self.assertEqual(get_docval(func, 'a')[0]['type'], ('int', str)) + + +class TestValidationParityHarness(TestCase): + """Test the parity harness itself on a known-equivalent function pair.""" + + def test_assert_validation_parity(self): + from hdmf.typing.testing import assert_validation_parity + from hdmf.utils import docval, getargs + + @docval({'name': 'name', 'type': str, 'doc': 'the name'}, + {'name': 'count', 'type': 'int', 'doc': 'how many', 'default': 1}, + is_method=False) + def old(**kwargs): + name, count = getargs('name', 'count', kwargs) + return (name, count) + + @validated + def new(name: str, count: Int = 1): + """N. + + Args: + name: the name + count: how many + """ + return (name, count) + + assert_validation_parity(old, new, [ + {'kwargs': {'name': 'a'}}, + {'kwargs': {'name': 'a', 'count': 5}}, + {'kwargs': {'name': 'a', 'count': np.int16(5)}}, + {'kwargs': {'name': 5}}, + {'kwargs': {'name': 'a', 'count': 'x'}}, + {'kwargs': {}}, + ])