diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml index ba50fd4..b529a99 100644 --- a/.github/workflows/python-test.yml +++ b/.github/workflows/python-test.yml @@ -5,6 +5,18 @@ on: - main pull_request: jobs: + codemeta-check: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Check codemeta.json version matches pyproject.toml + run: | + python3 -c " + import json, tomllib + cm = json.load(open('codemeta.json'))['version'] + py = tomllib.load(open('pyproject.toml','rb'))['project']['version'] + assert cm == py, f'codemeta.json version {cm} != pyproject {py}' + " build: runs-on: ${{ matrix.python-version == '3.8' && 'ubuntu-22.04' || 'ubuntu-latest' }} strategy: diff --git a/CITATION.cff b/CITATION.cff index 0517f29..ea76699 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -58,4 +58,4 @@ keywords: - EVI - Ontology - Provenance -license: MIT +license: Apache-2.0 diff --git a/codemeta.json b/codemeta.json new file mode 100644 index 0000000..08c1f50 --- /dev/null +++ b/codemeta.json @@ -0,0 +1,83 @@ +{ + "@context": "https://w3id.org/codemeta/3.0", + "@type": "SoftwareSourceCode", + "name": "fairscape-cli", + "description": "A utility for packaging objects and validating metadata for FAIRSCAPE. Provides RO-Crate creation and management, schema handling and validation, data import, build artifacts (HTML datasheets, evidence graphs), release management, and publishing to Fairscape, Dataverse, and DataCite.", + "version": "1.2.8", + "license": "https://spdx.org/licenses/Apache-2.0", + "codeRepository": "https://github.com/fairscape/fairscape-cli", + "issueTracker": "https://github.com/fairscape/fairscape-cli/issues", + "url": "https://fairscape.github.io/fairscape-cli/", + "downloadUrl": "https://pypi.org/project/fairscape-cli/", + "programmingLanguage": "Python", + "runtimePlatform": "Python >= 3.8", + "developmentStatus": "active", + "keywords": [ + "fairscape", + "reproducibility", + "FAIR", + "B2AI", + "CLI", + "RO-Crate" + ], + "author": [ + { + "@type": "Person", + "@id": "https://orcid.org/0000-0003-0384-8499", + "givenName": "Maxwell Adam", + "familyName": "Levinson", + "email": "mal8ch@virginia.edu", + "affiliation": { + "@type": "Organization", + "name": "University of Virginia" + } + }, + { + "@type": "Person", + "@id": "https://orcid.org/0000-0002-1103-3882", + "givenName": "Justin", + "familyName": "Niestroy", + "email": "jniestroy@gmail.com", + "affiliation": { + "@type": "Organization", + "name": "University of Virginia" + } + }, + { + "@type": "Person", + "@id": "https://orcid.org/0000-0003-4647-3877", + "givenName": "Sadnan", + "familyName": "Al Manir", + "email": "ma3xy@virginia.edu", + "affiliation": { + "@type": "Organization", + "name": "University of Virginia" + } + }, + { + "@type": "Person", + "@id": "https://orcid.org/0000-0003-4060-7360", + "givenName": "Timothy", + "familyName": "Clark", + "email": "twclark@virginia.edu", + "affiliation": { + "@type": "Organization", + "name": "University of Virginia" + } + } + ], + "softwareRequirements": [ + {"@type": "SoftwareApplication", "name": "click", "version": ">=8.1.7"}, + {"@type": "SoftwareApplication", "name": "pydantic", "version": ">=2.5.1"}, + {"@type": "SoftwareApplication", "name": "fairscape-models", "version": ">=1.2.0"}, + {"@type": "SoftwareApplication", "name": "jsonschema", "version": ">=4.20.0"}, + {"@type": "SoftwareApplication", "name": "frictionless", "version": ">=5.0,<6.0"}, + {"@type": "SoftwareApplication", "name": "rdflib"}, + {"@type": "SoftwareApplication", "name": "pandas"} + ], + "funder": { + "@type": "Organization", + "name": "National Institutes of Health" + }, + "funding": "NIH Bridge2AI: OT2OD032742 (Bridge2AI: Cell Maps for AI (CM4AI) Data Generation Project) and OT2OD032701 (Bridge2AI: Patient-Focused Collaborative Hospital Repository Uniting Standards (CHoRUS) for Equitable AI); Frederick Thomas Fund of the University of Virginia" +} diff --git a/pyproject.toml b/pyproject.toml index 4478aa8..7848ecd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,7 +39,7 @@ dependencies = [ "prettytable>=3.9.0", "jsonschema>=4.20.0", "sqids>=0.4.1", - "fairscape-models>=1.1.1", + "fairscape-models[schema-infer]>=1.2.0", "pyyaml", "h5py", "frictionless>=5.0,<6.0", diff --git a/src/fairscape_cli/commands/schema_commands.py b/src/fairscape_cli/commands/schema_commands.py index 607d11d..867d570 100644 --- a/src/fairscape_cli/commands/schema_commands.py +++ b/src/fairscape_cli/commands/schema_commands.py @@ -1,21 +1,21 @@ import click -import json -from prettytable import PrettyTable import pathlib +from prettytable import PrettyTable from pydantic import ( ValidationError ) -from typing import ( - Union, - Type -) +from fairscape_models.schema import Schema + +from fairscape_cli.config import DEFAULT_CONTEXT, DEFAULT_SCHEMA_TYPE from fairscape_cli.models import ReadROCrateMetadata, AppendCrate -from fairscape_cli.models.schema.tabular import ( - TabularValidationSchema, - HDF5ValidationSchema, +from fairscape_cli.models.schema import ( + infer_schema, + validate_schema, + load_schema, write_schema as WriteSchema, + generate_schema_guid, StringProperty, NumberProperty, IntegerProperty, @@ -52,18 +52,20 @@ def create_tabular_schema( """Initialize a Tabular Schema. """ try: - schema_model = TabularValidationSchema.model_validate({ + schema_model = Schema.model_validate({ + "@id": guid or generate_schema_guid(name), + "@context": DEFAULT_CONTEXT, + "@type": DEFAULT_SCHEMA_TYPE, "name": name, - "description":description, - "guid":guid, - "properties":{}, + "description": description, + "properties": {}, "required": [], - "header":header, + "header": header, "separator": separator }) except ValidationError as metadataError: - click.echo("ERROR Validating TabularValidationSchema") + click.echo("ERROR Validating Schema") for validationFailure in metadataError.errors(): click.echo(f"property: {validationFailure.get('loc')} \tmsg: {validationFailure.get('msg')}") ctx.exit(code=1) @@ -88,7 +90,7 @@ def add_property(): def add_property_string(ctx, name, index, description, value_url, pattern, schema_file): """Add a String Property to an existing Schema. """ - try: + try: stringPropertyModel = StringProperty.model_validate({ "name": name, "index": index, @@ -145,7 +147,7 @@ def add_property_number(ctx, name, index, description, maximum, minimum, value_u def add_property_boolean(ctx, name, index, description, value_url, schema_file): """Add a Boolean property to an existing Schema. """ - try: + try: booleanPropertyModel = BooleanProperty.model_validate({ "name": name, "index": index, @@ -209,7 +211,7 @@ def add_property_array(ctx, name, index, description, value_url, items_datatype, datatype_enum = DatatypeEnum(items_datatype) except Exception: print(f"ITEMS Datatype {items_datatype} invalid\n" + - "ITEMS must be oneOf 'boolean'|'object'|'string'|'number'|'integer'" + "ITEMS must be oneOf 'boolean'|'object'|'string'|'number'|'integer'" ) ctx.exit(code=1) @@ -226,70 +228,43 @@ def add_property_array(ctx, name, index, description, value_url, items_datatype, ) except ValidationError as metadataError: print("ERROR: MetadataValidationError") - for validationFailure in metadataError.errors(): + for validationFailure in metadataError.errors(): click.echo(f"property: {validationFailure.get('loc')} \tmsg: {validationFailure.get('msg')}") ctx.exit(code=1) ClickAppendProperty(ctx, schema_file, arrayPropertyModel, name) -def determine_schema_type(filepath: str) -> Type[Union[TabularValidationSchema, HDF5ValidationSchema]]: - """Determine which schema type to use based on file extension""" - ext = pathlib.Path(filepath).suffix.lower()[1:] - if ext in ('h5', 'hdf5'): - return HDF5ValidationSchema - elif ext in ('csv', 'tsv', 'parquet'): - return TabularValidationSchema - else: - raise ValueError(f"Unsupported file extension: {ext}") - @schema.command('validate') @click.option('--schema', type=str, required=True) @click.option('--data', type=str, required=True) @click.pass_context -def validate(ctx, schema, data): +def validate(ctx, schema, data): """Execute validation of a Schema against the provided data.""" if 'ark' not in schema: schema_path = pathlib.Path(schema) if not schema_path.exists(): click.echo(f"ERROR: Schema file at path {schema} does not exist") ctx.exit(1) - + data_path = pathlib.Path(data) if not data_path.exists(): click.echo(f"ERROR: Data file at path {data} does not exist") ctx.exit(1) try: - with open(schema) as f: - schema_json = json.load(f) - - schema_class = determine_schema_type(data) - validation_schema = schema_class.from_dict(schema_json) - - validation_errors = validation_schema.validate_file(data) + schema_model = load_schema(schema) + validation_errors = validate_schema(schema_model, data) if len(validation_errors) != 0: error_table = PrettyTable() - if isinstance(validation_schema, HDF5ValidationSchema): - error_table.field_names = ['path', 'error_type', 'failed_keyword', 'message'] - else: - error_table.field_names = ['row', 'error_type', 'failed_keyword', 'message'] - + error_table.field_names = ['location', 'error_type', 'failed_keyword', 'message'] for err in validation_errors: - if isinstance(validation_schema, HDF5ValidationSchema): - error_table.add_row([ - err.path, - err.type, - err.failed_keyword, - str(err.message) - ]) - else: - error_table.add_row([ - err.row, - err.type, - err.failed_keyword, - str(err.message) - ]) + error_table.add_row([ + err.location, + err.error_type, + err.failed_keyword, + str(err.message) + ]) print(error_table) ctx.exit(1) @@ -299,7 +274,7 @@ def validate(ctx, schema, data): except ValidationError as metadata_error: click.echo("Error with schema definition") - for validation_failure in metadata_error.errors(): + for validation_failure in metadata_error.errors(): click.echo(f"property: {validation_failure.get('loc')} \tmsg: {validation_failure.get('msg')}") ctx.exit(1) @@ -313,40 +288,37 @@ def validate(ctx, schema, data): @click.pass_context def infer_schema_rocrate(ctx, name, description, guid, input_file, rocrate_path, schema_file): """Infer a schema from a file and optionally append it to an RO-Crate. - + INPUT_FILE: File to infer schema from (CSV, TSV, Parquet, or HDF5) SCHEMA_FILE: Path to save the schema file """ try: - # Determine schema type and infer schema - schema_class = determine_schema_type(input_file) - schema_model = schema_class.infer_from_file( - input_file, - name, - description + schema_model = infer_schema( + input_file, + name, + description, + guid=guid or None ) - if guid: - schema_model.guid = guid WriteSchema(schema_model, schema_file) - + ext = pathlib.Path(input_file).suffix.lower()[1:] click.echo(f"Inferred Schema from {ext} file: {str(schema_file)}") - + # If RO-Crate path is provided, append the schema to it if rocrate_path: - + # Read the RO-Crate to verify it exists and is valid try: ReadROCrateMetadata(rocrate_path) except Exception as exc: click.echo(f"ERROR Reading ROCrate: {str(exc)}") ctx.exit(code=1) - + # Append to RO-Crate AppendCrate(cratePath=rocrate_path, elements=[schema_model]) click.echo(f"Added Schema to RO-Crate with ID: {schema_model.guid}") - + except ValueError as e: click.echo(f"Error with file type: {str(e)}") ctx.exit(code=1) @@ -365,104 +337,32 @@ def register_schema( schema_file: str, ): """Register a JSON Schema with the specified RO-Crate. - + ROCRATE-PATH: Path to the RO-Crate to add the schema to SCHEMA-FILE: Path to the schema JSON file """ try: - + try: ReadROCrateMetadata(rocrate_path) except Exception as exc: click.echo(f"ERROR Reading ROCrate: {str(exc)}") ctx.exit(code=1) - - # Read schema file - with open(schema_file, 'r') as f: - schema_data = json.load(f) - + try: - schema_model = TabularValidationSchema.from_dict(schema_data) - click.echo(f"Loaded schema as TabularValidationSchema") - except Exception as tabular_error: - # If that fails, try HDF5ValidationSchema - try: - schema_model = HDF5ValidationSchema.from_dict(schema_data) - click.echo(f"Loaded schema as HDF5ValidationSchema") - except Exception as hdf5_error: - click.echo(f"ERROR: Could not recognize schema format") - click.echo(f"TabularValidationSchema error: {str(tabular_error)}") - click.echo(f"HDF5ValidationSchema error: {str(hdf5_error)}") - ctx.exit(code=1) - + schema_model = load_schema(schema_file) + except Exception as load_error: + click.echo(f"ERROR: Could not recognize schema format") + click.echo(f"Schema error: {str(load_error)}") + ctx.exit(code=1) + AppendCrate(cratePath=rocrate_path, elements=[schema_model]) click.echo(f"Schema registered with ID: {schema_model.guid}") - + except Exception as exc: click.echo(f"ERROR: {str(exc)}") ctx.exit(code=1) -@schema.command('validate') -@click.option('--schema', type=str, required=True) -@click.option('--data', type=str, required=True) -@click.pass_context -def validate(ctx, schema, data): - """Execute validation of a Schema against the provided data.""" - if 'ark' not in schema: - schema_path = pathlib.Path(schema) - if not schema_path.exists(): - click.echo(f"ERROR: Schema file at path {schema} does not exist") - ctx.exit(1) - - data_path = pathlib.Path(data) - if not data_path.exists(): - click.echo(f"ERROR: Data file at path {data} does not exist") - ctx.exit(1) - - try: - with open(schema) as f: - schema_json = json.load(f) - - schema_class = determine_schema_type(data) - validation_schema = schema_class.from_dict(schema_json) - - validation_errors = validation_schema.validate_file(data) - - if len(validation_errors) != 0: - error_table = PrettyTable() - if isinstance(validation_schema, HDF5ValidationSchema): - error_table.field_names = ['path', 'error_type', 'failed_keyword', 'message'] - else: - error_table.field_names = ['row', 'error_type', 'failed_keyword', 'message'] - - for err in validation_errors: - if isinstance(validation_schema, HDF5ValidationSchema): - error_table.add_row([ - err.path, - err.type, - err.failed_keyword, - str(err.message) - ]) - else: - error_table.add_row([ - err.row, - err.type, - err.failed_keyword, - str(err.message) - ]) - - print(error_table) - ctx.exit(1) - else: - print('Validation Success') - ctx.exit(0) - - except ValidationError as metadata_error: - click.echo("Error with schema definition") - for validation_failure in metadata_error.errors(): - click.echo(f"property: {validation_failure.get('loc')} \tmsg: {validation_failure.get('msg')}") - ctx.exit(1) - # Placeholder for future RO-Crate structural validation # @validate_group.command('crate') # @click.argument('rocrate-path', type=click.Path(exists=True, path_type=pathlib.Path)) @@ -470,4 +370,4 @@ def validate(ctx, schema, data): # """Validate the structure and metadata of an RO-Crate.""" # # Implementation using RO-Crate-py or custom checks # click.echo(f"Validating RO-Crate at {rocrate_path} (Not implemented yet)") -# pass \ No newline at end of file +# pass diff --git a/src/fairscape_cli/models/pep.py b/src/fairscape_cli/models/pep.py index 05db8c2..0fb0753 100644 --- a/src/fairscape_cli/models/pep.py +++ b/src/fairscape_cli/models/pep.py @@ -11,7 +11,7 @@ CopyToROCrate ) from fairscape_cli.models.guid_utils import GenerateDatetimeSquid -from fairscape_cli.models.schema.tabular import TabularValidationSchema +from fairscape_cli.models.schema import infer_schema from fairscape_cli.config import NAAN @@ -159,7 +159,7 @@ def _add_sample_path_to_rocrate(self, schema = None try: - schema = TabularValidationSchema.infer_from_file( + schema = infer_schema( str(source_path), f"Schema for {dataset_name}", f"Automatically inferred schema for {dataset_name}" @@ -218,7 +218,7 @@ def _register_subsample_path(self, schema = None try: - schema = TabularValidationSchema.infer_from_file( + schema = infer_schema( str(source_path), f"Schema for {dataset_name}", f"Automatically inferred schema for {dataset_name}" diff --git a/src/fairscape_cli/models/schema/__init__.py b/src/fairscape_cli/models/schema/__init__.py index e69de29..99dc230 100644 --- a/src/fairscape_cli/models/schema/__init__.py +++ b/src/fairscape_cli/models/schema/__init__.py @@ -0,0 +1,72 @@ +from typing import List, Optional + +from fairscape_models.schema import Schema, Property + +from fairscape_cli.models.schema.core import ( + SchemaHandler, + ValidationErrorRecord, + frictionless_type_to_json_schema, + generate_schema_guid, + SOURCE_TYPE_KEY, + register_handler, + get_handler, + supported_extensions, + load_schema, + write_schema, + normalize_schema_document, +) +from fairscape_cli.models.schema.properties import ( + DatatypeEnum, + Items, + NullProperty, + StringProperty, + ArrayProperty, + BooleanProperty, + NumberProperty, + IntegerProperty, + AppendProperty, + ClickAppendProperty, +) + +# Importing handlers registers the built-in csv/tsv, hdf5, and parquet handlers. +import fairscape_cli.models.schema.handlers # noqa: F401,E402 + + +def infer_schema(filepath: str, name: str, description: str, + guid: Optional[str] = None) -> Schema: + """Infer a canonical Schema from a data file, dispatching by extension.""" + return get_handler(filepath).infer(filepath, name, description, guid=guid) + + +def validate_schema(schema: Schema, filepath: str) -> List[ValidationErrorRecord]: + """Validate a data file against a canonical Schema, dispatching by extension.""" + return get_handler(filepath).validate(schema, filepath) + + +__all__ = [ + 'Schema', + 'Property', + 'SchemaHandler', + 'ValidationErrorRecord', + 'frictionless_type_to_json_schema', + 'generate_schema_guid', + 'SOURCE_TYPE_KEY', + 'register_handler', + 'get_handler', + 'supported_extensions', + 'load_schema', + 'write_schema', + 'normalize_schema_document', + 'infer_schema', + 'validate_schema', + 'DatatypeEnum', + 'Items', + 'NullProperty', + 'StringProperty', + 'ArrayProperty', + 'BooleanProperty', + 'NumberProperty', + 'IntegerProperty', + 'AppendProperty', + 'ClickAppendProperty', +] diff --git a/src/fairscape_cli/models/schema/core.py b/src/fairscape_cli/models/schema/core.py new file mode 100644 index 0000000..468a1dd --- /dev/null +++ b/src/fairscape_cli/models/schema/core.py @@ -0,0 +1,252 @@ +import json +import pathlib +import re +from abc import ABC, abstractmethod +from typing import ClassVar, Dict, List, Optional, Tuple, Union + +from pydantic import BaseModel + +from fairscape_models.schema import Schema, Property + +from fairscape_cli.config import ( + DEFAULT_CONTEXT, + DEFAULT_SCHEMA_TYPE, + NAAN, +) +from fairscape_cli.models.guid_utils import GenerateDatetimeSquid +from fairscape_cli.utils.serialization import model_dump_pruned, write_json_atomic + +CANONICAL_TYPES = {'integer', 'number', 'string', 'array', 'boolean', 'object'} +SOURCE_TYPE_KEY = 'source-type' + + +def frictionless_type_to_json_schema(field_type: str) -> str: + """Convert Frictionless types to JSON Schema types""" + type_mapping = { + 'string': 'string', + 'integer': 'integer', + 'number': 'number', + 'boolean': 'boolean', + 'date': 'string', + 'datetime': 'string', + 'year': 'integer', + 'yearmonth': 'string', + 'duration': 'string', + 'geopoint': 'array', + 'geojson': 'object', + 'array': 'array', + 'object': 'object', + 'time': 'string' + } + return type_mapping.get(field_type, 'string') + + +def generate_schema_guid(name: str) -> str: + """Generate a unique identifier for a schema""" + prefix = f"schema-{name.lower().replace(' ', '-')}" + return f"ark:{NAAN}/{prefix}-{GenerateDatetimeSquid()}" + + +# --------------------------------------------------------------------------- # +# Validation error record + frictionless adapter +# --------------------------------------------------------------------------- # + +class ValidationErrorRecord(BaseModel): + message: str + failed_keyword: str = "error" + error_type: str = "ValidationError" + field: Optional[str] = None + row: Optional[int] = None + path: Optional[str] = None + + @property + def location(self) -> str: + parts = [ + part for part in ( + self.path, + f"row {self.row}" if self.row is not None else None, + self.field, + ) if part + ] + return " / ".join(parts) or "-" + + +def frictionless_error_to_record(error, path: Optional[str] = None) -> ValidationErrorRecord: + """Convert a frictionless error object to a ValidationErrorRecord. + + Frictionless 5.x exposes row/field attributes under both snake_case and + camelCase depending on error class and minor version, so check both. + """ + row = getattr(error, 'row_number', None) + if row is None: + row = getattr(error, 'rowNumber', None) + field = getattr(error, 'field_name', None) + if field is None: + field = getattr(error, 'fieldName', None) + return ValidationErrorRecord( + message=error.message, + row=row, + field=field, + failed_keyword=getattr(error, 'type', 'error') or 'error', + path=path, + ) + + +# --------------------------------------------------------------------------- # +# Handler contract + registry +# --------------------------------------------------------------------------- # + +class SchemaHandler(ABC): + """Contract for per-format schema inference and validation. + + Implementations may use any machinery internally, but infer() must return + the canonical fairscape_models Schema and validate() must consume it. + Register implementations with @register_handler('ext', ...). + """ + + extensions: ClassVar[Tuple[str, ...]] = () + + @abstractmethod + def infer(self, filepath: str, name: str, description: str, + guid: Optional[str] = None) -> Schema: + """Infer a canonical Schema from a data file.""" + + @abstractmethod + def validate(self, schema: Schema, filepath: str) -> List[ValidationErrorRecord]: + """Validate a data file against a canonical Schema.""" + + def new_schema(self, *, name: str, description: str, + properties: Dict[str, Property], required: List[str], + guid: Optional[str] = None, separator: str = ",", + header: bool = True, **extra) -> Schema: + return Schema.model_validate({ + "@id": guid or generate_schema_guid(name), + "@context": DEFAULT_CONTEXT, + "@type": DEFAULT_SCHEMA_TYPE, + "name": name, + "description": description, + "properties": properties, + "required": required, + "separator": separator, + "header": header, + **extra, + }) + + +_HANDLERS: Dict[str, SchemaHandler] = {} + + +def register_handler(*extensions: str): + """Class decorator: instantiate the handler and map each extension to it. + + Third parties can add formats by subclassing SchemaHandler and decorating + with @register_handler('myext') in code that imports fairscape_cli. + """ + def decorator(cls): + handler = cls() + for ext in extensions: + _HANDLERS[ext.lower().lstrip('.')] = handler + return cls + return decorator + + +def get_handler(filepath: str) -> SchemaHandler: + ext = pathlib.Path(filepath).suffix.lower().lstrip('.') + try: + return _HANDLERS[ext] + except KeyError: + raise ValueError( + f"Unsupported file extension '{ext}'. " + f"Supported extensions: {', '.join(supported_extensions())}" + ) + + +def supported_extensions() -> List[str]: + return sorted(_HANDLERS) + + +# --------------------------------------------------------------------------- # +# Loading (with backward-compat normalization) + writing +# --------------------------------------------------------------------------- # + +# Same index pattern the canonical Property model accepts +_INDEX_PATTERN = re.compile(r'^\d+$|^-?\d+::|^-?\d+::-?\d+$|^::-?\d+') + + +def _valid_index(value) -> bool: + if isinstance(value, int): + return True + if isinstance(value, str): + return bool(_INDEX_PATTERN.match(value)) + return False + + +def normalize_property(name: str, prop: dict, index_fallback: int) -> dict: + """Normalize a single property dict to canonical Property shape. + + Handles legacy HDF5 schema JSON where each dataset entry is a full + TabularValidationSchema dump (nested 'properties' but no 'index'), and + legacy non-canonical type strings like 'datetime'/'year'. + """ + if not isinstance(prop, dict): + return prop + + nested = prop.get('properties') + if isinstance(nested, dict) and not _valid_index(prop.get('index')): + # Legacy HDF5 dataset entry: rebuild as an object Property, dropping + # the schema-level wrapper keys the old dump carried. + return { + 'type': 'object', + 'index': index_fallback, + 'description': prop.get('description') or f"Dataset at {name}", + 'hdf5-path': name, + 'properties': { + child_name: normalize_property(child_name, child, i) + for i, (child_name, child) in enumerate(nested.items()) + }, + } + + normalized = dict(prop) + prop_type = normalized.get('type') + if prop_type is None: + normalized['type'] = 'string' + elif prop_type not in CANONICAL_TYPES: + normalized.setdefault(SOURCE_TYPE_KEY, prop_type) + normalized['type'] = frictionless_type_to_json_schema(prop_type) + + if not _valid_index(normalized.get('index')): + normalized['index'] = index_fallback + if not normalized.get('description'): + normalized['description'] = f"Column {name}" + + if isinstance(nested, dict): + normalized['properties'] = { + child_name: normalize_property(child_name, child, i) + for i, (child_name, child) in enumerate(nested.items()) + } + return normalized + + +def normalize_schema_document(data: dict) -> dict: + """Normalize a schema JSON document so legacy files validate canonically.""" + data = dict(data) + data.setdefault('@type', DEFAULT_SCHEMA_TYPE) + data.setdefault('separator', ',') + data.setdefault('header', True) + data['properties'] = { + name: normalize_property(name, prop, i) + for i, (name, prop) in enumerate(data.get('properties', {}).items()) + } + return data + + +def load_schema(path: Union[str, pathlib.Path]) -> Schema: + """Read a schema JSON file into the canonical Schema model.""" + with open(path) as f: + data = json.load(f) + return Schema.model_validate(normalize_schema_document(data)) + + +def write_schema(schema: Schema, output_file: Union[str, pathlib.Path]) -> None: + """Write a canonical Schema to a JSON file.""" + write_json_atomic(output_file, model_dump_pruned(schema, by_alias=True)) diff --git a/src/fairscape_cli/models/schema/handlers/__init__.py b/src/fairscape_cli/models/schema/handlers/__init__.py new file mode 100644 index 0000000..c50d9b6 --- /dev/null +++ b/src/fairscape_cli/models/schema/handlers/__init__.py @@ -0,0 +1,6 @@ +# Importing these modules registers the built-in handlers as a side effect. +from fairscape_cli.models.schema.handlers import tabular +from fairscape_cli.models.schema.handlers import hdf5 +from fairscape_cli.models.schema.handlers import parquet + +__all__ = ['tabular', 'hdf5', 'parquet'] diff --git a/src/fairscape_cli/models/schema/handlers/hdf5.py b/src/fairscape_cli/models/schema/handlers/hdf5.py new file mode 100644 index 0000000..aa0a92c --- /dev/null +++ b/src/fairscape_cli/models/schema/handlers/hdf5.py @@ -0,0 +1,131 @@ +from typing import List, Optional + +import h5py +import pandas as pd +from frictionless import Resource, describe + +from fairscape_models.schema import Schema, Property + +from fairscape_cli.models.schema.core import ( + SchemaHandler, + ValidationErrorRecord, + frictionless_error_to_record, + frictionless_type_to_json_schema, + register_handler, + SOURCE_TYPE_KEY, +) +from fairscape_cli.models.schema.handlers.tabular import build_frictionless_schema + + +def dataset_to_dataframe(dataset: h5py.Dataset) -> pd.DataFrame: + """Convert an HDF5 dataset to a pandas DataFrame""" + data = dataset[()] + + if dataset.dtype.fields: # Structured array + return pd.DataFrame(data) + elif len(dataset.shape) > 1: # Multi-dimensional array + n_cols = dataset.shape[1] + columns = [f"column_{i}" for i in range(n_cols)] + return pd.DataFrame(data, columns=columns) + else: # 1D array + return pd.DataFrame(data, columns=['value']) + + +@register_handler('h5', 'hdf5') +class HDF5Handler(SchemaHandler): + extensions = ('h5', 'hdf5') + + def infer(self, filepath: str, name: str, description: str, + guid: Optional[str] = None) -> Schema: + properties = {} + + with h5py.File(filepath, 'r') as f: + def process_group(group, parent_path=""): + for key, item in group.items(): + path = f"{parent_path}/{key}" if parent_path else key + + if isinstance(item, h5py.Dataset): + try: + df = dataset_to_dataframe(item) + resource = describe(df) + + columns = {} + for i, field in enumerate(resource.schema.fields): + json_schema_type = frictionless_type_to_json_schema(field.type) + extra = {} + if json_schema_type != field.type: + extra[SOURCE_TYPE_KEY] = field.type + columns[field.name] = Property( + type=json_schema_type, + description=field.description or f"Column {field.name}", + index=i, + **extra, + ) + + # The int index is a placeholder required by the + # canonical Property index validator; validation + # keys datasets on the path, never this index. + properties[path] = Property( + type='object', + index=len(properties), + description=f"Dataset at {path}", + properties=columns, + **{'hdf5-path': path}, + ) + + except Exception as e: + print(f"Warning: Could not process dataset {path}: {str(e)}") + + elif isinstance(item, h5py.Group): + process_group(item, path) + + process_group(f) + + return self.new_schema( + name=name, + description=description, + guid=guid, + properties=properties, + required=list(properties.keys()), + ) + + def validate(self, schema: Schema, filepath: str) -> List[ValidationErrorRecord]: + errors = [] + + with h5py.File(filepath, 'r') as f: + for path, prop in schema.properties.items(): + if prop.type != 'object' or not prop.properties: + continue + try: + dataset = f[path] + if isinstance(dataset, h5py.Dataset): + df = dataset_to_dataframe(dataset) + frictionless_schema = build_frictionless_schema(prop.properties) + resource = Resource(data=df, schema=frictionless_schema) + report = resource.validate() + + for task in report.tasks: + for error in task.errors: + # Skip string type errors: h5py returns bytes + # for string data, which frictionless flags as + # a type mismatch against string fields. + if (hasattr(error, 'type') and error.type == 'type-error' and + hasattr(error, 'note') and 'type is "string' in error.note): + continue + + errors.append(frictionless_error_to_record(error, path=path)) + + except KeyError: + errors.append(ValidationErrorRecord( + message=f"Dataset {path} not found", + failed_keyword="required", + path=path, + )) + except Exception as e: + errors.append(ValidationErrorRecord( + message=f"Error validating dataset {path}: {str(e)}", + failed_keyword="format", + path=path, + )) + + return errors diff --git a/src/fairscape_cli/models/schema/handlers/parquet.py b/src/fairscape_cli/models/schema/handlers/parquet.py new file mode 100644 index 0000000..be77c73 --- /dev/null +++ b/src/fairscape_cli/models/schema/handlers/parquet.py @@ -0,0 +1,111 @@ +from typing import List, Optional + +import pyarrow as pa +import pyarrow.parquet as pq + +from fairscape_models.schema import Schema, Property + +from fairscape_cli.models.schema.core import ( + SchemaHandler, + ValidationErrorRecord, + register_handler, + SOURCE_TYPE_KEY, +) + + +def arrow_type_to_json_schema(arrow_type: pa.DataType) -> str: + """Map an Arrow type to one of the six canonical JSON Schema types.""" + if pa.types.is_boolean(arrow_type): + return 'boolean' + if pa.types.is_integer(arrow_type): + return 'integer' + if pa.types.is_floating(arrow_type) or pa.types.is_decimal(arrow_type): + return 'number' + if (pa.types.is_list(arrow_type) or pa.types.is_large_list(arrow_type) + or pa.types.is_fixed_size_list(arrow_type)): + return 'array' + if pa.types.is_struct(arrow_type) or pa.types.is_map(arrow_type): + return 'object' + # string/binary/timestamp/date/time/duration/dictionary/... + return 'string' + + +@register_handler('parquet') +class ParquetHandler(SchemaHandler): + """Native Parquet handler. + + Reads the file's embedded Arrow schema directly — no full-table read and + no pandas round-trip — so exact source types (int64 vs float64, + timestamp[us] vs date32) are preserved in each Property's 'source-type'. + Validation compares the embedded schema against the declared one without + scanning rows. + """ + extensions = ('parquet',) + + def infer(self, filepath: str, name: str, description: str, + guid: Optional[str] = None) -> Schema: + arrow_schema = pq.read_schema(filepath) + + properties = {} + required_fields = [] + + for i, field in enumerate(arrow_schema): + properties[field.name] = Property( + type=arrow_type_to_json_schema(field.type), + description=f"Column {field.name}", + index=i, + **{SOURCE_TYPE_KEY: str(field.type)}, + ) + required_fields.append(field.name) + + return self.new_schema( + name=name, + description=description, + guid=guid, + properties=properties, + required=required_fields, + ) + + def validate(self, schema: Schema, filepath: str) -> List[ValidationErrorRecord]: + arrow_schema = pq.read_schema(filepath) + file_fields = {field.name: field for field in arrow_schema} + + errors = [] + + for name, prop in schema.properties.items(): + field = file_fields.get(name) + if field is None: + errors.append(ValidationErrorRecord( + message=f"Column {name} not found in parquet file", + failed_keyword="required", + field=name, + )) + continue + + actual_source = str(field.type) + actual_canonical = arrow_type_to_json_schema(field.type) + declared_source = (prop.model_extra or {}).get(SOURCE_TYPE_KEY) + + if declared_source is not None and declared_source == actual_source: + continue + # Fall back to comparing canonical categories so hand-authored + # schemas (no source-type) or compatible physical-type rewrites + # still validate. + if prop.type != actual_canonical: + declared = declared_source or prop.type + errors.append(ValidationErrorRecord( + message=f"Column {name} type mismatch: schema declares {declared}, file has {actual_source}", + failed_keyword="type", + field=name, + )) + + if schema.additionalProperties is False: + for name in file_fields: + if name not in schema.properties: + errors.append(ValidationErrorRecord( + message=f"Column {name} present in parquet file but not declared in schema", + failed_keyword="additionalProperties", + field=name, + )) + + return errors diff --git a/src/fairscape_cli/models/schema/handlers/tabular.py b/src/fairscape_cli/models/schema/handlers/tabular.py new file mode 100644 index 0000000..e4e0ffb --- /dev/null +++ b/src/fairscape_cli/models/schema/handlers/tabular.py @@ -0,0 +1,166 @@ +import pathlib +from typing import Dict, List, Optional + +from frictionless import ( + Schema as FrictionlessSchema, + Resource, + Dialect, + Report, + describe, + fields, + formats, +) + +from fairscape_models.schema import Schema, Property + +from fairscape_cli.models.schema.core import ( + SchemaHandler, + ValidationErrorRecord, + frictionless_error_to_record, + frictionless_type_to_json_schema, + register_handler, + SOURCE_TYPE_KEY, +) + +_TYPE_TO_FRICTIONLESS_FIELD = { + 'string': fields.StringField, + 'integer': fields.IntegerField, + 'number': fields.NumberField, + 'boolean': fields.BooleanField, +} + + +def _get_either(prop_details: dict, *keys): + for key in keys: + if prop_details.get(key) is not None: + return prop_details[key] + return None + + +def build_frictionless_schema(properties: Dict[str, Property]) -> FrictionlessSchema: + """Rebuild a frictionless Schema from canonical Properties for row validation.""" + properties_input = { + name: prop.model_dump(by_alias=True, exclude_none=True) + for name, prop in properties.items() + } + + frictionless_schema_obj = FrictionlessSchema() + + sorted_prop_items = [] + spanning_array_prop_name = None + spanning_array_prop_details = None + + for name, prop_details in properties_input.items(): + index_val = prop_details.get("index") + if prop_details.get("type") == "array" and isinstance(index_val, str) and "::" in index_val: + if spanning_array_prop_name is not None: + raise ValueError("Multiple spanning array properties (index: 'X::') are not supported.") + spanning_array_prop_name = name + spanning_array_prop_details = prop_details + elif isinstance(index_val, int): + sorted_prop_items.append((name, prop_details, index_val)) + else: + sorted_prop_items.append((name, prop_details, float('inf'))) + + sorted_prop_items.sort(key=lambda x: x[2]) + + for name, prop_details, _ in sorted_prop_items: + field_class = _TYPE_TO_FRICTIONLESS_FIELD.get(prop_details.get('type', 'string'), fields.StringField) + + constraints = {} + if 'minimum' in prop_details: constraints['minimum'] = prop_details['minimum'] + if 'maximum' in prop_details: constraints['maximum'] = prop_details['maximum'] + if 'pattern' in prop_details: constraints['pattern'] = prop_details['pattern'] + if 'minLength' in prop_details: constraints['minLength'] = prop_details['minLength'] + if 'maxLength' in prop_details: constraints['maxLength'] = prop_details['maxLength'] + + if prop_details.get('type') == 'array': + field = fields.ArrayField(name=name, description=prop_details.get('description', ''), constraints=constraints) + else: + field = field_class(name=name, description=prop_details.get('description', ''), constraints=constraints) + frictionless_schema_obj.add_field(field) + + if spanning_array_prop_name and spanning_array_prop_details: + prop_name_original = spanning_array_prop_name + details = spanning_array_prop_details + item_details = details.get('items', {}) + item_type = item_details.get('type', 'number') + item_field_class = _TYPE_TO_FRICTIONLESS_FIELD.get(item_type, fields.NumberField) + + # min/max item counts appear as 'min-items'/'max-items' (canonical alias) + # or 'minItems'/'maxItems' (CLI property models) depending on origin + num_items = _get_either(details, 'min-items', 'minItems') + max_items = _get_either(details, 'max-items', 'maxItems') + if num_items is None or num_items != max_items: + raise ValueError(f"Spanning array '{prop_name_original}' must have equal and defined minItems and maxItems.") + + for i in range(num_items): + field_name_for_frictionless = f"{prop_name_original}_{i}" # e.g., embed_0, embed_1, ... + field = item_field_class(name=field_name_for_frictionless, description=f"Element {i} of {prop_name_original}") + frictionless_schema_obj.add_field(field) + + return frictionless_schema_obj + + +@register_handler('csv', 'tsv') +class TabularHandler(SchemaHandler): + extensions = ('csv', 'tsv') + + def infer(self, filepath: str, name: str, description: str, + guid: Optional[str] = None) -> Schema: + ext = pathlib.Path(filepath).suffix.lower().lstrip('.') + separator = '\t' if ext == 'tsv' else ',' + + resource = describe(filepath) + + properties = {} + required_fields = [] + + for i, field in enumerate(resource.schema.fields): + json_schema_type = frictionless_type_to_json_schema(field.type) + extra = {} + if json_schema_type != field.type: + extra[SOURCE_TYPE_KEY] = field.type + + properties[field.name] = Property( + type=json_schema_type, + description=field.description or f"Column {field.name}", + index=i, + **extra, + ) + required_fields.append(field.name) + + return self.new_schema( + name=name, + description=description, + guid=guid, + properties=properties, + required=required_fields, + separator=separator, + header=True, + ) + + def validate(self, schema: Schema, filepath: str) -> List[ValidationErrorRecord]: + frictionless_schema = build_frictionless_schema(schema.properties) + + file_dialect = Dialect() + file_dialect.header = schema.header if schema.header is not None else True + + if schema.separator: + csv_control = formats.csv.CsvControl(delimiter=schema.separator) + file_dialect.add_control(csv_control) + + resource = Resource( + path=filepath, + schema=frictionless_schema, + dialect=file_dialect, + ) + + report: Report = resource.validate() + errors_list = [] + if not report.valid: + for task in report.tasks: + for error_detail in task.errors: + errors_list.append(frictionless_error_to_record(error_detail)) + + return errors_list diff --git a/src/fairscape_cli/models/schema/properties.py b/src/fairscape_cli/models/schema/properties.py new file mode 100644 index 0000000..784c938 --- /dev/null +++ b/src/fairscape_cli/models/schema/properties.py @@ -0,0 +1,143 @@ +import pathlib +from enum import Enum +from typing import Literal, Optional, Union + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + model_validator, +) + +from fairscape_models.schema import Property + +from fairscape_cli.models.schema.core import load_schema, write_schema +from fairscape_cli.models.schema.utils import ( + PropertyNameException, + ColumnIndexException, +) + + +class DatatypeEnum(str, Enum): + NULL = "null" + BOOLEAN = "boolean" + STRING = "string" + NUMBER = "number" + INTEGER = "integer" + ARRAY = "array" + + +class Items(BaseModel): + model_config = ConfigDict( + populate_by_name = True, + use_enum_values=True + ) + datatype: DatatypeEnum = Field(alias="type") + + +class BaseProperty(BaseModel): + description: str = Field(description="description of field") + model_config = ConfigDict(populate_by_name = True) + index: Union[int,str] = Field(description="index of the column for this value") + valueURL: Optional[str] = Field(default=None) + + +class NullProperty(BaseProperty): + datatype: Literal['null'] = Field(alias="type", default='null') + index: int + + +class StringProperty(BaseProperty): + datatype: Literal['string'] = Field(alias="type") + pattern: Optional[str] = Field(description="Regex pattern to execute against values", default=None) + maxLength: Optional[int] = Field(description="Inclusive maximum length for string values", default=None) + minLength: Optional[int] = Field(description="Inclusive minimum length for string values", default=None) + index: int + + +class ArrayProperty(BaseProperty): + datatype: Literal['array'] = Field(alias="type") + maxItems: Optional[int] = Field(description="max items in array, validation fails if length is greater than this value", default=None) + minItems: Optional[int] = Field(description="min items in array, validation fails if lenght is shorter than this value", default=None) + uniqueItems: Optional[bool] = Field() + index: str + items: Items + + +class BooleanProperty(BaseProperty): + datatype: Literal['boolean'] = Field(alias="type") + index: int + + +class NumberProperty(BaseProperty): + datatype: Literal['number'] = Field(alias="type") + maximum: Optional[float] = Field(description="Inclusive Upper Limit for Values", default=None) + minimum: Optional[float] = Field(description="Inclusive Lower Limit for Values", default=None) + index: int + + @model_validator(mode='after') + def check_max_min(self) -> 'NumberProperty': + minimum = self.minimum + maximum = self.maximum + + if maximum is not None and minimum is not None: + if maximum == minimum: + raise ValueError('NumberProperty attribute minimum != maximum') + elif maximum < minimum: + raise ValueError('NumberProperty attribute maximum !< minimum') + return self + + +class IntegerProperty(BaseProperty): + datatype: Literal['integer'] = Field(alias="type") + maximum: Optional[int] = Field(description="Inclusive Upper Limit for Values", default=None) + minimum: Optional[int] = Field(description="Inclusive Lower Limit for Values", default=None) + index: int + + @model_validator(mode='after') + def check_max_min(self) -> 'IntegerProperty': + minimum = self.minimum + maximum = self.maximum + + if maximum is not None and minimum is not None: + if maximum == minimum: + raise ValueError('IntegerProperty attribute minimum != maximum') + elif maximum < minimum: + raise ValueError('IntegerProperty attribute maximum !< minimum') + return self + + +def AppendProperty(schemaFilepath: str, propertyInstance, propertyName: str) -> None: + schemaPath = pathlib.Path(schemaFilepath) + if not schemaPath.exists(): + raise FileNotFoundError(f"Schema file {schemaFilepath} does not exist") + + schemaModel = load_schema(schemaPath) + + if propertyName in schemaModel.properties: + raise PropertyNameException(propertyName) + + canonical_property = Property.model_validate( + propertyInstance.model_dump(by_alias=True, exclude_none=True) + ) + schemaModel.properties[propertyName] = canonical_property + schemaModel.required.append(propertyName) + + write_schema(schemaModel, schemaPath) + + +def ClickAppendProperty(ctx, schemaFile, propertyModel, name): + try: + # append the property to the schema file + AppendProperty(schemaFile, propertyModel, name) + print(f"Added Property\tname: {name}\ttype: {propertyModel.datatype}") + ctx.exit(code=0) + + except ColumnIndexException as indexException: + print("ERROR: ColumnIndexError") + print(str(indexException)) + ctx.exit(code=1) + except PropertyNameException as propertyException: + print("ERROR: PropertyNameError") + print(str(propertyException)) + ctx.exit(code=1) diff --git a/src/fairscape_cli/models/schema/tabular.py b/src/fairscape_cli/models/schema/tabular.py deleted file mode 100644 index f34d5f4..0000000 --- a/src/fairscape_cli/models/schema/tabular.py +++ /dev/null @@ -1,582 +0,0 @@ -import pathlib -import os -import json -import pyarrow.parquet as pq -import pandas as pd -import h5py -from datetime import datetime -from enum import Enum -from pydantic import ( - BaseModel, - ConfigDict, - Field, - ValidationError, - model_validator -) -from typing import ( - Dict, - List, - Optional, - Literal, - Union -) -from frictionless import Schema, Resource, fields, Dialect, Report, describe, formats - - -from fairscape_cli.models.schema.utils import ( - PropertyNameException, - ColumnIndexException, -) - -from fairscape_cli.config import ( - DEFAULT_CONTEXT, - DEFAULT_SCHEMA_TYPE, - NAAN, -) -from fairscape_cli.utils.serialization import model_dump_pruned - -class FileType(str, Enum): - CSV = "csv" - TSV = "tsv" - PARQUET = "parquet" - - @classmethod - def from_extension(cls, filepath: str) -> 'FileType': - ext = pathlib.Path(filepath).suffix.lower()[1:] - if ext == 'parquet': - return cls.PARQUET - elif ext == 'tsv': - return cls.TSV - elif ext == 'csv': - return cls.CSV - else: - raise ValueError(f"Unsupported file extension: {ext}") - -class ValidationError(BaseModel): - message: str - row: Optional[int] = None - field: Optional[str] = None - type: str = "ValidationError" - failed_keyword: str - path: Optional[str] = None - -class DatatypeEnum(str, Enum): - NULL = "null" - BOOLEAN = "boolean" - STRING = "string" - NUMBER = "number" - INTEGER = "integer" - ARRAY = "array" - -class Items(BaseModel): - model_config = ConfigDict( - populate_by_name = True, - use_enum_values=True - ) - datatype: DatatypeEnum = Field(alias="type") - -class BaseProperty(BaseModel): - description: str = Field(description="description of field") - model_config = ConfigDict(populate_by_name = True) - index: Union[int,str] = Field(description="index of the column for this value") - valueURL: Optional[str] = Field(default=None) - -class NullProperty(BaseProperty): - datatype: Literal['null'] = Field(alias="type", default='null') - index: int - -class StringProperty(BaseProperty): - datatype: Literal['string'] = Field(alias="type") - pattern: Optional[str] = Field(description="Regex pattern to execute against values", default=None) - maxLength: Optional[int] = Field(description="Inclusive maximum length for string values", default=None) - minLength: Optional[int] = Field(description="Inclusive minimum length for string values", default=None) - index: int - -class ArrayProperty(BaseProperty): - datatype: Literal['array'] = Field(alias="type") - maxItems: Optional[int] = Field(description="max items in array, validation fails if length is greater than this value", default=None) - minItems: Optional[int] = Field(description="min items in array, validation fails if lenght is shorter than this value", default=None) - uniqueItems: Optional[bool] = Field() - index: str - items: Items - -class BooleanProperty(BaseProperty): - datatype: Literal['boolean'] = Field(alias="type") - index: int - -class NumberProperty(BaseProperty): - datatype: Literal['number'] = Field(alias="type") - maximum: Optional[float] = Field(description="Inclusive Upper Limit for Values", default=None) - minimum: Optional[float] = Field(description="Inclusive Lower Limit for Values", default=None) - index: int - - @model_validator(mode='after') - def check_max_min(self) -> 'NumberProperty': - minimum = self.minimum - maximum = self.maximum - - if maximum is not None and minimum is not None: - if maximum == minimum: - raise ValueError('NumberProperty attribute minimum != maximum') - elif maximum < minimum: - raise ValueError('NumberProperty attribute maximum !< minimum') - return self - -class IntegerProperty(BaseProperty): - datatype: Literal['integer'] = Field(alias="type") - maximum: Optional[int] = Field(description="Inclusive Upper Limit for Values", default=None) - minimum: Optional[int] = Field(description="Inclusive Lower Limit for Values", default=None) - index: int - - @model_validator(mode='after') - def check_max_min(self) -> 'IntegerProperty': - minimum = self.minimum - maximum = self.maximum - - if maximum is not None and minimum is not None: - if maximum == minimum: - raise ValueError('IntegerProperty attribute minimum != maximum') - elif maximum < minimum: - raise ValueError('IntegerProperty attribute maximum !< minimum') - return self - -def frictionless_type_to_json_schema(field_type: str) -> str: - """Convert Frictionless types to JSON Schema types""" - type_mapping = { - 'string': 'string', - 'integer': 'integer', - 'number': 'number', - 'boolean': 'boolean', - 'date': 'string', - 'datetime': 'string', - 'year': 'integer', - 'yearmonth': 'string', - 'duration': 'string', - 'geopoint': 'array', - 'geojson': 'object', - 'array': 'array', - 'object': 'object', - 'time': 'string' - } - return type_mapping.get(field_type, 'string') - -class TabularValidationSchema(BaseModel): - model_config = ConfigDict(populate_by_name=True) - - guid: Optional[str] = Field(alias="@id", default=None) - context: Optional[Dict] = Field(default=DEFAULT_CONTEXT, alias="@context") - metadataType: Optional[str] = Field(default=DEFAULT_SCHEMA_TYPE, alias="@type") - schema_version: str = Field(default="https://json-schema.org/draft/2020-12/schema", alias="$schema") - name: str - description: str - datatype: str = Field(default="object", alias="type") - separator: str = Field(description="Field separator for the file") - header: bool = Field(description="Do files of this schema have a header row", default=True) - required: List[str] = Field(default=[]) - properties: Dict[str, Dict] = Field(default={}) - additionalProperties: bool = Field(default=True) - - # Store the frictionless schema - _frictionless_schema: Optional[Schema] = None - - def generate_guid(self) -> str: - """Generate a unique identifier for the schema""" - if self.guid is None: - prefix = f"schema-{self.name.lower().replace(' ', '-')}" - timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") - self.guid = f"ark:{NAAN}/{prefix}-{timestamp}" - return self.guid - - @model_validator(mode='after') - def generate_all_guids(self) -> 'TabularValidationSchema': - """Generate GUIDs for this schema and any nested schemas""" - self.generate_guid() - return self - - @classmethod - def infer_from_file(cls, filepath: str, name: str, description: str, include_min_max: bool = False) -> 'TabularValidationSchema': - """Infer schema from a file using Frictionless""" - file_type = FileType.from_extension(filepath) - separator = '\t' if file_type == FileType.TSV else ',' - - # if parquet file need to read in file with pyarrow to pandas - if file_type == FileType.PARQUET: - pd_table = pq.read_table(filepath).to_pandas() - resource = describe(pd_table) - - else: - resource = describe(filepath) - - properties = {} - required_fields = [] - - for i, field in enumerate(resource.schema.fields): - json_schema_type = frictionless_type_to_json_schema(field.type) - - property_def = { - "type": json_schema_type, - "description": field.description or f"Column {field.name}", - "index": i - } - - properties[field.name] = property_def - required_fields.append(field.name) - - # Create our schema instance - schema = cls( - name=name, - description=description, - separator=separator, - header=True, - properties=properties, - required=required_fields - ) - - # Store the frictionless schema for validation - schema._frictionless_schema = resource.schema - return schema - - def validate_file(self, filepath: str) -> List[ValidationError]: - if not self._frictionless_schema: - raise ValueError("Frictionless schema not properly initialized. Call from_dict or infer_from_file.") - - file_dialect = Dialect() - - file_dialect.header = self.header - - if self.separator: - csv_control = formats.csv.CsvControl(delimiter=self.separator) - file_dialect.add_control(csv_control) - - resource = Resource( - path=filepath, - schema=self._frictionless_schema, - dialect=file_dialect - ) - - report: Report = resource.validate() - errors_list = [] - if not report.valid: - for task in report.tasks: # Changed from report.errors - for error_detail in task.errors: - validation_error_model = ValidationError( - message=error_detail.message, - row=error_detail.row_number if hasattr(error_detail, 'row_number') else None, - field=error_detail.field_name if hasattr(error_detail, 'field_name') else None, - failed_keyword=error_detail.type if hasattr(error_detail, 'type') else 'error' - ) - errors_list.append(validation_error_model) - - return errors_list - - def to_dict(self) -> dict: - """Convert the schema to a dictionary format""" - return self.model_dump(by_alias=True, exclude={'_frictionless_schema'}) - - @classmethod - def from_dict(cls, data: dict) -> 'TabularValidationSchema': - """Create a schema instance from a dictionary""" - - properties_input = data.get('properties', {}) - required_fields_input = data.get('required', []) - header_setting = data.get('header', True) - separator_setting = data.get('separator', ',') - - frictionless_schema_obj = Schema() - - type_to_frictionless_field = { - 'string': fields.StringField, - 'integer': fields.IntegerField, - 'number': fields.NumberField, - 'boolean': fields.BooleanField, - } - - sorted_prop_items = [] - spanning_array_prop_name = None - spanning_array_prop_details = None - - for name, prop_details in properties_input.items(): - index_val = prop_details.get("index") - if prop_details.get("type") == "array" and isinstance(index_val, str) and "::" in index_val: - if spanning_array_prop_name is not None: - raise ValueError("Multiple spanning array properties (index: 'X::') are not supported.") - spanning_array_prop_name = name - spanning_array_prop_details = prop_details - elif isinstance(index_val, int): - sorted_prop_items.append((name, prop_details, index_val)) - else: - sorted_prop_items.append((name, prop_details, float('inf'))) - - sorted_prop_items.sort(key=lambda x: x[2]) - - for name, prop_details, _ in sorted_prop_items: - if name == spanning_array_prop_name: - continue - - field_class = type_to_frictionless_field.get(prop_details.get('type', 'string'), fields.StringField) - - constraints = {} - if 'minimum' in prop_details: constraints['minimum'] = prop_details['minimum'] - if 'maximum' in prop_details: constraints['maximum'] = prop_details['maximum'] - if 'pattern' in prop_details: constraints['pattern'] = prop_details['pattern'] - if 'minLength' in prop_details: constraints['minLength'] = prop_details['minLength'] - if 'maxLength' in prop_details: constraints['maxLength'] = prop_details['maxLength'] - - if prop_details.get('type') == 'array': - field = fields.ArrayField(name=name, description=prop_details.get('description', ''), constraints=constraints) - else: - field = field_class(name=name, description=prop_details.get('description', ''), constraints=constraints) - frictionless_schema_obj.add_field(field) - - if spanning_array_prop_name and spanning_array_prop_details: - prop_name_original = spanning_array_prop_name - details = spanning_array_prop_details - item_details = details.get('items', {}) - item_type = item_details.get('type', 'number') - item_field_class = type_to_frictionless_field.get(item_type, fields.NumberField) - - num_items = details.get('minItems') - if num_items is None or num_items != details.get('maxItems'): - raise ValueError(f"Spanning array '{prop_name_original}' must have equal and defined minItems and maxItems.") - - for i in range(num_items): - field_name_for_frictionless = f"{prop_name_original}_{i}" # e.g., embed_0, embed_1, ... - field = item_field_class(name=field_name_for_frictionless, description=f"Element {i} of {prop_name_original}") - frictionless_schema_obj.add_field(field) - - schema_instance_data = {k: v for k, v in data.items() if k not in ['properties', 'required', 'header', 'separator']} - - schema_instance = cls( - **schema_instance_data, - properties=properties_input, - required=required_fields_input, - header=header_setting, - separator=separator_setting - ) - schema_instance._frictionless_schema = frictionless_schema_obj - return schema_instance - -class HDF5ValidationSchema(BaseModel): - guid: Optional[str] = Field(alias="@id", default=None) - context: Optional[Dict] = Field(default=DEFAULT_CONTEXT, alias="@context") - name: str - description: str - properties: Dict[str, TabularValidationSchema] = Field(default={}) - required: List[str] = Field(default=[]) - - def generate_guid(self) -> str: - """Generate a unique identifier for the schema""" - if self.guid is None: - prefix = f"schema-{self.name.lower().replace(' ', '-')}" - timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") - self.guid = f"ark:{NAAN}/{prefix}-{timestamp}" - return self.guid - - @model_validator(mode='after') - def generate_all_guids(self) -> 'HDF5ValidationSchema': - """Generate GUIDs for this schema and any nested schemas""" - self.generate_guid() - return self - - @staticmethod - def dataset_to_dataframe(dataset: h5py.Dataset) -> pd.DataFrame: - """Convert an HDF5 dataset to a pandas DataFrame""" - data = dataset[()] - - if dataset.dtype.fields: # Structured array - return pd.DataFrame(data) - elif len(dataset.shape) > 1: # Multi-dimensional array - n_cols = dataset.shape[1] - columns = [f"column_{i}" for i in range(n_cols)] - return pd.DataFrame(data, columns=columns) - else: # 1D array - return pd.DataFrame(data, columns=['value']) - - @classmethod - def infer_from_file(cls, filepath: str, name: str, description: str) -> 'HDF5ValidationSchema': - """Infer schema from an HDF5 file""" - schema = cls( - name=name, - description=description - ) - properties = {} - - with h5py.File(filepath, 'r') as f: - def process_group(group, parent_path=""): - for key, item in group.items(): - path = f"{parent_path}/{key}" if parent_path else key - - if isinstance(item, h5py.Dataset): - try: - df = cls.dataset_to_dataframe(item) - resource = describe(df) - - tabular_schema = TabularValidationSchema( - name=f"{name}_{path.replace('/', '_')}", - description=f"Dataset at {path}", - separator=",", - header=True, - properties={}, - required=[], - context=None - ) - - tabular_schema._frictionless_schema = resource.schema - - for i, field in enumerate(resource.schema.fields): - property_def = { - "type": field.type, - "description": field.description or f"Column {field.name}", - "index": i - } - - tabular_schema.properties[field.name] = property_def - tabular_schema.required.append(field.name) - - properties[path] = tabular_schema - - except Exception as e: - print(f"Warning: Could not process dataset {path}: {str(e)}") - - elif isinstance(item, h5py.Group): - process_group(item, path) - - process_group(f) - schema.properties = properties - schema.required = list(properties.keys()) - - return schema - - def validate_file(self, filepath: str) -> List[ValidationError]: - """Validate an HDF5 file against the schema""" - errors = [] - - with h5py.File(filepath, 'r') as f: - for path, schema in self.properties.items(): - try: - dataset = f[path] - if isinstance(dataset, h5py.Dataset): - df = self.dataset_to_dataframe(dataset) - resource = Resource(data=df, schema=schema._frictionless_schema) - report = resource.validate() - - for task in report.tasks: - for error in task.errors: - # Skip string type errors - if (hasattr(error, 'type') and error.type == 'type-error' and - hasattr(error, 'note') and 'type is "string' in error.note): - continue - - validation_error = ValidationError( - message=error.message, - row=error.rowNumber if hasattr(error, 'rowNumber') else None, - field=error.fieldName if hasattr(error, 'fieldName') else None, - type="ValidationError", - failed_keyword=error.type if hasattr(error, 'type') else "error", - path=path - ) - errors.append(validation_error) - - except KeyError: - errors.append(ValidationError( - message=f"Dataset {path} not found", - type="ValidationError", - failed_keyword="required", - path=path - )) - except Exception as e: - errors.append(ValidationError( - message=f"Error validating dataset {path}: {str(e)}", - type="ValidationError", - failed_keyword="format", - path=path - )) - - return errors - - def to_dict(self) -> dict: - """Convert the schema to a dictionary format including all fields""" - return self.model_dump(by_alias=True) - - @classmethod - def from_dict(cls, data: dict) -> 'HDF5ValidationSchema': - """Create a schema instance from a dictionary""" - properties = { - path: TabularValidationSchema.from_dict(schema_dict) - for path, schema_dict in data.get('properties', {}).items() - } - - return cls( - name=data['name'], - description=data['description'], - properties=properties, - required=data.get('required', []) - ) - -def write_schema(schema: TabularValidationSchema, output_file: str): - """Write a schema to a file""" - schema_dict = model_dump_pruned(schema, by_alias=True) - - with open(output_file, 'w') as f: - json.dump(schema_dict, f, indent=2) - -def AppendProperty(schemaFilepath: str, propertyInstance, propertyName: str) -> None: - # check that schemaFile exists - schemaPath = pathlib.Path(schemaFilepath) - if not schemaPath.exists(): - raise Exception - - with schemaPath.open("r+") as schemaFile: - schemaFileContents = schemaFile.read() - schemaJson = json.loads(schemaFileContents) - - schemaModel = TabularValidationSchema.model_validate(schemaJson) - - if propertyName in [key for key in schemaModel.properties.keys()]: - raise PropertyNameException(propertyName) - - schema_indicies = [val['index'] for val in schemaModel.properties.values()] - - schemaModel.properties[propertyName] = propertyInstance - schemaModel.required.append(propertyName) - schemaJson = json.dumps(schemaModel.model_dump(by_alias=True, exclude_none=True), indent=2) - - # overwrite file contents - schemaFile.seek(0) - schemaFile.write(schemaJson) - -def ClickAppendProperty(ctx, schemaFile, propertyModel, name): - try: - # append the property to the - AppendProperty(schemaFile, propertyModel, name) - print(f"Added Property\tname: {name}\ttype: {propertyModel.datatype}") - ctx.exit(code=0) - - except ColumnIndexException as indexException: - print("ERROR: ColumnIndexError") - print(str(indexException)) - ctx.exit(code=1) - except PropertyNameException as propertyException: - print("ERROR: PropertyNameError") - print(str(propertyException)) - ctx.exit(code=1) - -def ReadSchemaGithub(schemaURI: str) -> TabularValidationSchema: - pass - -def ReadSchemaFairscape(schemaArk: str) -> TabularValidationSchema: - pass - -def ReadSchemaLocal(schemaFile: str) -> TabularValidationSchema: - """ Helper function for reading the schema and marshaling into the pydantic model - """ - schemaPath = pathlib.Path(schemaFile) - - # read the schema - with schemaPath.open("r") as inputSchema: - inputSchemaData = inputSchema.read() - schemaJson = json.loads(inputSchemaData) - - # load the model into - tabularSchema = TabularValidationSchema.model_validate(schemaJson) - return tabularSchema diff --git a/tests/commands/schema/test_schema_handlers.py b/tests/commands/schema/test_schema_handlers.py new file mode 100644 index 0000000..dc901e1 --- /dev/null +++ b/tests/commands/schema/test_schema_handlers.py @@ -0,0 +1,245 @@ +import json +import pathlib + +import numpy as np +import pyarrow as pa +import pyarrow.parquet as pq +import h5py +import pandas as pd +import pytest + +from fairscape_cli.__main__ import cli as fairscape_cli_app + + +@pytest.fixture +def sample_parquet(tmp_path: pathlib.Path) -> pathlib.Path: + """Parquet file exercising exact type fidelity (int64, float64, string, timestamp).""" + table = pa.table({ + "count": pa.array([1, 2, 3], type=pa.int64()), + "value": pa.array([1.5, 2.5, 3.5], type=pa.float64()), + "label": pa.array(["a", "b", "c"]), + "ts": pa.array(pd.to_datetime(["2024-01-01", "2024-01-02", "2024-01-03"]), + type=pa.timestamp("us")), + }) + path = tmp_path / "sample.parquet" + pq.write_table(table, path) + return path + + +@pytest.fixture +def sample_h5(tmp_path: pathlib.Path) -> pathlib.Path: + """HDF5 file with a root 1-D dataset and a nested structured-array dataset.""" + path = tmp_path / "sample.h5" + with h5py.File(path, "w") as f: + f.create_dataset("root_data", data=np.arange(10.0)) + grp = f.create_group("grp") + arr = np.array([(1, 2.5), (2, 3.5)], dtype=[("id", "i8"), ("score", "f8")]) + grp.create_dataset("ds", data=arr) + return path + + +class TestParquetHandler: + def test_infer_preserves_exact_types(self, runner, tmp_path, sample_parquet): + schema_path = tmp_path / "s_pq.json" + result = runner.invoke(fairscape_cli_app, [ + "schema", "infer", + "--name", "Parquet Schema", + "--description", "Schema inferred from a parquet file for testing", + str(sample_parquet), str(schema_path), + ]) + assert result.exit_code == 0, result.output + + data = json.loads(schema_path.read_text()) + props = data["properties"] + assert props["count"]["type"] == "integer" + assert props["count"]["source-type"] == "int64" + assert props["value"]["type"] == "number" + assert props["value"]["source-type"] == "double" + assert props["label"]["type"] == "string" + # timestamp collapses to the canonical 'string' but records the exact source + assert props["ts"]["type"] == "string" + assert props["ts"]["source-type"] == "timestamp[us]" + + def test_validate_success(self, runner, tmp_path, sample_parquet): + schema_path = tmp_path / "s_pq.json" + runner.invoke(fairscape_cli_app, [ + "schema", "infer", "--name", "PQ", "--description", "parquet schema test", + str(sample_parquet), str(schema_path), + ]) + result = runner.invoke(fairscape_cli_app, [ + "schema", "validate", "--schema", str(schema_path), "--data", str(sample_parquet), + ]) + assert result.exit_code == 0, result.output + assert "Validation Success" in result.output + + def test_validate_type_and_missing_column(self, runner, tmp_path, sample_parquet): + schema_path = tmp_path / "s_pq.json" + runner.invoke(fairscape_cli_app, [ + "schema", "infer", "--name", "PQ", "--description", "parquet schema test", + str(sample_parquet), str(schema_path), + ]) + data = json.loads(schema_path.read_text()) + # break the count column's declared source type + add a nonexistent column + data["properties"]["count"]["type"] = "number" + data["properties"]["count"]["source-type"] = "float64" + data["properties"]["missing_col"] = {"type": "string", "index": 9, "description": "not present"} + schema_path.write_text(json.dumps(data)) + + result = runner.invoke(fairscape_cli_app, [ + "schema", "validate", "--schema", str(schema_path), "--data", str(sample_parquet), + ]) + assert result.exit_code != 0 + assert "type" in result.output + assert "required" in result.output + + +class TestHDF5Handler: + def test_infer_maps_datasets_to_object_properties(self, runner, tmp_path, sample_h5): + schema_path = tmp_path / "s_h5.json" + result = runner.invoke(fairscape_cli_app, [ + "schema", "infer", + "--name", "HDF5 Schema", + "--description", "Schema inferred from an hdf5 file for testing", + str(sample_h5), str(schema_path), + ]) + assert result.exit_code == 0, result.output + + data = json.loads(schema_path.read_text()) + props = data["properties"] + assert "grp/ds" in props + assert "root_data" in props + ds = props["grp/ds"] + assert ds["type"] == "object" + assert ds["hdf5-path"] == "grp/ds" + assert ds["properties"]["id"]["type"] == "integer" + assert ds["properties"]["score"]["type"] == "number" + # every nested column type is one of the six canonical types + canonical = {"integer", "number", "string", "array", "boolean", "object"} + for dataset_prop in props.values(): + for col in dataset_prop["properties"].values(): + assert col["type"] in canonical + + def test_validate_success(self, runner, tmp_path, sample_h5): + schema_path = tmp_path / "s_h5.json" + runner.invoke(fairscape_cli_app, [ + "schema", "infer", "--name", "H5", "--description", "hdf5 schema test", + str(sample_h5), str(schema_path), + ]) + result = runner.invoke(fairscape_cli_app, [ + "schema", "validate", "--schema", str(schema_path), "--data", str(sample_h5), + ]) + assert result.exit_code == 0, result.output + assert "Validation Success" in result.output + + def test_validate_missing_dataset(self, runner, tmp_path, sample_h5): + schema_path = tmp_path / "s_h5.json" + runner.invoke(fairscape_cli_app, [ + "schema", "infer", "--name", "H5", "--description", "hdf5 schema test", + str(sample_h5), str(schema_path), + ]) + data = json.loads(schema_path.read_text()) + data["properties"]["grp/nope"] = json.loads(json.dumps(data["properties"]["grp/ds"])) + data["properties"]["grp/nope"]["hdf5-path"] = "grp/nope" + schema_path.write_text(json.dumps(data)) + + result = runner.invoke(fairscape_cli_app, [ + "schema", "validate", "--schema", str(schema_path), "--data", str(sample_h5), + ]) + assert result.exit_code != 0 + assert "required" in result.output + assert "grp/nope" in result.output + + def test_add_to_crate(self, runner, tmp_path, sample_h5): + schema_path = tmp_path / "s_h5.json" + runner.invoke(fairscape_cli_app, [ + "schema", "infer", "--name", "H5 Crate Schema", + "--description", "hdf5 schema for crate integration testing", + str(sample_h5), str(schema_path), + ]) + crate_dir = tmp_path / "crate" + runner.invoke(fairscape_cli_app, [ + "rocrate", "create", str(crate_dir), + "--name", "Test Crate", "--organization-name", "Org", + "--project-name", "Proj", + "--description", "A test crate for hdf5 schema integration testing", + "--keywords", "testing", + ]) + # Previously HDF5 schemas lacked @type and failed crate validation. + result = runner.invoke(fairscape_cli_app, [ + "schema", "add-to-crate", str(crate_dir), str(schema_path), + ]) + assert result.exit_code == 0, result.output + + metadata = json.loads((crate_dir / "ro-crate-metadata.json").read_text()) + schema_entity = next( + item for item in metadata["@graph"] + if item.get("@type") == "EVI:Schema" + ) + assert schema_entity["name"] == "H5 Crate Schema" + assert "fairscapeVersion" in schema_entity + + +class TestLegacyCompat: + def test_legacy_hdf5_schema_loads_validates_and_registers(self, runner, tmp_path, sample_h5): + legacy = { + "@id": "ark:59852/schema-legacy-hdf5", + "name": "legacy hdf5", + "description": "legacy format hdf5 schema for backward-compat testing", + "required": ["root_data", "grp/ds"], + "properties": { + "root_data": { + "@id": "ark:59852/schema-sub1", "name": "legacy_root_data", + "description": "Dataset at root_data", "type": "object", + "separator": ",", "header": True, "additionalProperties": True, + "required": ["value"], + "properties": {"value": {"type": "number", "description": "Column value", "index": 0}}, + }, + "grp/ds": { + "@id": "ark:59852/schema-sub2", "name": "legacy_grp_ds", + "description": "Dataset at grp/ds", "type": "object", + "separator": ",", "header": True, "additionalProperties": True, + "required": ["id", "score"], + "properties": { + # non-canonical legacy frictionless type that must be + # normalized to a canonical type ('year' -> 'integer') + "id": {"type": "year", "description": "Column id", "index": 0}, + "score": {"type": "number", "description": "Column score", "index": 1}, + }, + }, + }, + } + schema_path = tmp_path / "legacy_hdf5_schema.json" + schema_path.write_text(json.dumps(legacy, indent=2)) + + # the legacy non-canonical 'year' type must normalize to 'integer' on load + from fairscape_cli.models.schema import load_schema + normalized = load_schema(schema_path) + id_prop = normalized.properties["grp/ds"].properties["id"] + assert id_prop.type == "integer" + assert (id_prop.model_extra or {}).get("source-type") == "year" + + validate_result = runner.invoke(fairscape_cli_app, [ + "schema", "validate", "--schema", str(schema_path), "--data", str(sample_h5), + ]) + assert validate_result.exit_code == 0, validate_result.output + + crate_dir = tmp_path / "crate" + runner.invoke(fairscape_cli_app, [ + "rocrate", "create", str(crate_dir), + "--name", "Legacy Crate", "--organization-name", "Org", + "--project-name", "Proj", + "--description", "A test crate for legacy schema backward compat testing", + "--keywords", "testing", + ]) + add_result = runner.invoke(fairscape_cli_app, [ + "schema", "add-to-crate", str(crate_dir), str(schema_path), + ]) + assert add_result.exit_code == 0, add_result.output + + +class TestValidateCommandRegistration: + def test_validate_registered_once(self, runner): + """The duplicate 'validate' definition was removed; it must appear once.""" + result = runner.invoke(fairscape_cli_app, ["schema", "--help"]) + assert result.exit_code == 0 + assert result.output.count("validate") == 1