ShEx Validation API

Overview

The ShEx validation module provides RDF data validation against ShEx (Shape Expression) schemas. It's designed primarily for validating Wikidata entities against EntitySchemas but supports any RDF data and ShEx schema combination.

Current implementations: Wikidata EntitySchema validation, local file validation
Future implementations: Additional RDF graph sources, streaming validation

Quick Start

from gkc import ShexValidator

# Validate Wikidata item against EntitySchema
validator = ShexValidator(qid='Q14708404', eid='E502')
result = validator.check()

if result.is_valid():
    print("✓ Validation passed")
else:
    print("✗ Validation failed")

Classes

ShexValidator

ShEx Validator: Validate RDF data against ShEx schemas.

Validates Wikidata entities or local RDF data against EntitySchemas (ShEx format). Supports multiple input sources: Wikidata entities, local files, or text strings.

Plain meaning: Check if data matches schema structure and rules.

Example

Validate a Wikidata item against an EntitySchema

validator = ShexValidator(qid='Q42', eid='E502') result = validator.check() if result.is_valid(): ... print("Validation passed!")

Use local schema file

validator = ShexValidator( ... qid='Q42', ... schema_file='schema.shex' ... ) validator.check()

Use RDF text directly

validator = ShexValidator( ... rdf_text=my_rdf_data, ... schema_text=my_schema ... ) validator.check()

Source code in gkc/shex.py

class ShexValidator:
    """
    ShEx Validator: Validate RDF data against ShEx schemas.

    Validates Wikidata entities or local RDF data against EntitySchemas (ShEx format).
    Supports multiple input sources: Wikidata entities, local files, or text strings.

    Plain meaning: Check if data matches schema structure and rules.

    Example:
        >>> # Validate a Wikidata item against an EntitySchema
        >>> validator = ShexValidator(qid='Q42', eid='E502')
        >>> result = validator.check()
        >>> if result.is_valid():
        ...     print("Validation passed!")

        >>> # Use local schema file
        >>> validator = ShexValidator(
        ...     qid='Q42',
        ...     schema_file='schema.shex'
        ... )
        >>> validator.check()

        >>> # Use RDF text directly
        >>> validator = ShexValidator(
        ...     rdf_text=my_rdf_data,
        ...     schema_text=my_schema
        ... )
        >>> validator.check()
    """

    def __init__(
        self,
        qid: Optional[str] = None,
        eid: Optional[str] = None,
        user_agent: Optional[str] = None,
        schema_text: Optional[str] = None,
        schema_file: Optional[str] = None,
        rdf_text: Optional[str] = None,
        rdf_file: Optional[str] = None,
    ):
        """
        Initialize the ShEx validator.

        Args:
            qid: Wikidata entity ID (e.g., 'Q42'). Optional if rdf_text or
                rdf_file provided.
            eid: EntitySchema ID for Wikidata schema (e.g., 'E502').
                Optional if schema_text or schema_file provided.
            user_agent: Custom user agent for Wikidata requests.
            schema_text: ShEx schema as ShExC string (alternative to eid).
            schema_file: Path to file containing ShEx schema (alternative to eid).
            rdf_text: RDF data as a string (alternative to qid).
            rdf_file: Path to file containing RDF data (alternative to qid).
        """
        self.qid = qid
        self.eid = eid
        self.user_agent = user_agent
        self.schema_text = schema_text
        self.schema_file = schema_file
        self.rdf_text = rdf_text
        self.rdf_file = rdf_file

        self._schema: Optional[str] = None
        self._rdf: Optional[str] = None
        self.results = None

    def load_specification(self) -> "ShexValidator":
        """
        Load the ShEx schema specification from configured source.

        Tries sources in order: schema_text, schema_file, eid (fetch from Wikidata).

        Returns:
            Self for method chaining

        Raises:
            ShexValidationError: If no valid schema source or loading fails
        """
        try:
            if self.schema_text:
                self._schema = self.schema_text
            elif self.schema_file:
                schema_path = Path(self.schema_file)
                if not schema_path.exists():
                    raise ShexValidationError(
                        f"Schema file not found: {self.schema_file}"
                    )
                self._schema = schema_path.read_text(encoding="utf-8")
            elif self.eid:
                self._schema = fetch_schema_specification(self.eid, self.user_agent)
            else:
                raise ShexValidationError(
                    "No schema source provided. "
                    "Specify eid, schema_text, or schema_file."
                )
        except RuntimeError as e:
            raise ShexValidationError(f"Failed to load schema: {str(e)}") from e
        except OSError as e:
            msg = f"Failed to read schema file: {str(e)}"
            raise ShexValidationError(msg) from e

        return self

    def load_rdf(self) -> "ShexValidator":
        """
        Load RDF data from configured source.

        Tries sources in order: rdf_text, rdf_file, qid (from Wikidata).

        Returns:
            Self for method chaining

        Raises:
            ShexValidationError: If no valid RDF source or loading fails
        """
        try:
            if self.rdf_text:
                self._rdf = self.rdf_text
            elif self.rdf_file:
                rdf_path = Path(self.rdf_file)
                if not rdf_path.exists():
                    msg = f"RDF file not found: {self.rdf_file}"
                    raise ShexValidationError(msg)
                self._rdf = rdf_path.read_text(encoding="utf-8")
            elif self.qid:
                self._rdf = fetch_entity_rdf(
                    self.qid, format="ttl", user_agent=self.user_agent
                )
            else:
                raise ShexValidationError(
                    "No RDF source provided. Specify qid, rdf_text, or rdf_file."
                )
        except RuntimeError as e:
            raise ShexValidationError(f"Failed to load RDF: {str(e)}") from e
        except OSError as e:
            raise ShexValidationError(f"Failed to read RDF file: {str(e)}") from e

        return self

    def passes_inspection(self) -> bool:
        """
        Check if validation passed (alias for is_valid).

        Returns:
            True if validation passed, False otherwise

        Raises:
            ShexValidationError: If check() hasn't been called yet
        """
        return self.is_valid()

    def evaluate(self) -> "ShexValidator":
        """
        Evaluate RDF data against the ShEx schema specification.

        Must call load_specification() and load_rdf() first, or use check().

        Returns:
            Self with results populated

        Raises:
            ShexValidationError: If evaluation fails or data not loaded
        """
        if self._schema is None:
            raise ShexValidationError(
                "Schema not loaded. Call load_specification() first or use check()."
            )
        if self._rdf is None:
            raise ShexValidationError(
                "RDF data not loaded. Call load_rdf() first or use check()."
            )

        # Determine focus node
        focus = None
        if self.qid:
            focus = get_entity_uri(self.qid)

        try:
            self.results = ShExEvaluator(
                rdf=self._rdf, schema=self._schema, focus=focus
            ).evaluate()
        except Exception as e:
            msg = f"ShEx evaluation failed: {str(e)}"
            raise ShexValidationError(msg) from e

        return self

    def check(self) -> "ShexValidator":
        """
        Validate: Load schema, load RDF, and evaluate in one call.

        This is the main entry point for validation. It loads schema and data
        from configured sources, then performs the validation.

        Returns:
            Self with results populated

        Example:
            >>> validator = ShexValidator(qid='Q42', eid='E502')
            >>> validator.check()
            >>> if validator.is_valid():
            ...     print("Validation passed!")
        """
        self.load_specification()
        self.load_rdf()
        self.evaluate()
        return self

    def is_valid(self) -> bool:
        """
        Check if validation passed.

        Returns:
            True if validation passed, False otherwise

        Raises:
            ShexValidationError: If check() hasn't been called yet
        """
        if self.results is None:
            msg = "No validation results. Call check() first."
            raise ShexValidationError(msg)

        # Handle mocked results (for testing)
        if isinstance(self.results, bool):
            return self.results

        # PyShEx returns results as a list of EvaluationResult objects
        # When validation succeeds, reason contains matching triples
        # When validation fails, reason contains error messages like
        # "Node: ... not in value set"
        # If no focus is specified, PyShEx tests all nodes;
        # we need at least one success
        if not self.results:
            return False

        # Check if at least one result succeeded (no error indicators)
        for result in self.results:
            reason = result.reason or ""
            # Common failure indicators in PyShEx error messages
            has_error = any(
                indicator in reason
                for indicator in [
                    "not in value set",
                    "does not match",
                    "Constraint violation",
                    "No matching",
                    "Failed to",
                ]
            )
            if not has_error:
                return True

        return False

    def __repr__(self) -> str:
        """String representation of validator."""
        parts = []
        if self.qid:
            parts.append(f"qid={self.qid!r}")
        if self.eid:
            parts.append(f"eid={self.eid!r}")
        if self.rdf_file:
            parts.append(f"rdf_file={self.rdf_file!r}")
        if self.schema_file:
            parts.append(f"schema_file={self.schema_file!r}")

        params = ", ".join(parts) if parts else ""
        return f"ShexValidator({params})"

__init__(qid=None, eid=None, user_agent=None, schema_text=None, schema_file=None, rdf_text=None, rdf_file=None)

Initialize the ShEx validator.

Parameters:

Name	Type	Description	Default
qid	Optional[str]	Wikidata entity ID (e.g., 'Q42'). Optional if rdf_text or rdf_file provided.	None
eid	Optional[str]	EntitySchema ID for Wikidata schema (e.g., 'E502'). Optional if schema_text or schema_file provided.	None
user_agent	Optional[str]	Custom user agent for Wikidata requests.	None
schema_text	Optional[str]	ShEx schema as ShExC string (alternative to eid).	None
schema_file	Optional[str]	Path to file containing ShEx schema (alternative to eid).	None
rdf_text	Optional[str]	RDF data as a string (alternative to qid).	None
rdf_file	Optional[str]	Path to file containing RDF data (alternative to qid).	None

Source code in gkc/shex.py

def __init__(
    self,
    qid: Optional[str] = None,
    eid: Optional[str] = None,
    user_agent: Optional[str] = None,
    schema_text: Optional[str] = None,
    schema_file: Optional[str] = None,
    rdf_text: Optional[str] = None,
    rdf_file: Optional[str] = None,
):
    """
    Initialize the ShEx validator.

    Args:
        qid: Wikidata entity ID (e.g., 'Q42'). Optional if rdf_text or
            rdf_file provided.
        eid: EntitySchema ID for Wikidata schema (e.g., 'E502').
            Optional if schema_text or schema_file provided.
        user_agent: Custom user agent for Wikidata requests.
        schema_text: ShEx schema as ShExC string (alternative to eid).
        schema_file: Path to file containing ShEx schema (alternative to eid).
        rdf_text: RDF data as a string (alternative to qid).
        rdf_file: Path to file containing RDF data (alternative to qid).
    """
    self.qid = qid
    self.eid = eid
    self.user_agent = user_agent
    self.schema_text = schema_text
    self.schema_file = schema_file
    self.rdf_text = rdf_text
    self.rdf_file = rdf_file

    self._schema: Optional[str] = None
    self._rdf: Optional[str] = None
    self.results = None

__repr__()

String representation of validator.

Source code in gkc/shex.py

def __repr__(self) -> str:
    """String representation of validator."""
    parts = []
    if self.qid:
        parts.append(f"qid={self.qid!r}")
    if self.eid:
        parts.append(f"eid={self.eid!r}")
    if self.rdf_file:
        parts.append(f"rdf_file={self.rdf_file!r}")
    if self.schema_file:
        parts.append(f"schema_file={self.schema_file!r}")

    params = ", ".join(parts) if parts else ""
    return f"ShexValidator({params})"

check()

Validate: Load schema, load RDF, and evaluate in one call.

This is the main entry point for validation. It loads schema and data from configured sources, then performs the validation.

Returns:

Type	Description
ShexValidator	Self with results populated

Example

validator = ShexValidator(qid='Q42', eid='E502') validator.check() if validator.is_valid(): ... print("Validation passed!")

Source code in gkc/shex.py

def check(self) -> "ShexValidator":
    """
    Validate: Load schema, load RDF, and evaluate in one call.

    This is the main entry point for validation. It loads schema and data
    from configured sources, then performs the validation.

    Returns:
        Self with results populated

    Example:
        >>> validator = ShexValidator(qid='Q42', eid='E502')
        >>> validator.check()
        >>> if validator.is_valid():
        ...     print("Validation passed!")
    """
    self.load_specification()
    self.load_rdf()
    self.evaluate()
    return self

evaluate()

Evaluate RDF data against the ShEx schema specification.

Must call load_specification() and load_rdf() first, or use check().

Returns:

Type	Description
ShexValidator	Self with results populated

Raises:

Type	Description
ShexValidationError	If evaluation fails or data not loaded

Source code in gkc/shex.py

def evaluate(self) -> "ShexValidator":
    """
    Evaluate RDF data against the ShEx schema specification.

    Must call load_specification() and load_rdf() first, or use check().

    Returns:
        Self with results populated

    Raises:
        ShexValidationError: If evaluation fails or data not loaded
    """
    if self._schema is None:
        raise ShexValidationError(
            "Schema not loaded. Call load_specification() first or use check()."
        )
    if self._rdf is None:
        raise ShexValidationError(
            "RDF data not loaded. Call load_rdf() first or use check()."
        )

    # Determine focus node
    focus = None
    if self.qid:
        focus = get_entity_uri(self.qid)

    try:
        self.results = ShExEvaluator(
            rdf=self._rdf, schema=self._schema, focus=focus
        ).evaluate()
    except Exception as e:
        msg = f"ShEx evaluation failed: {str(e)}"
        raise ShexValidationError(msg) from e

    return self

is_valid()

Check if validation passed.

Returns:

Type	Description
bool	True if validation passed, False otherwise

Raises:

Type	Description
ShexValidationError	If check() hasn't been called yet

Source code in gkc/shex.py

def is_valid(self) -> bool:
    """
    Check if validation passed.

    Returns:
        True if validation passed, False otherwise

    Raises:
        ShexValidationError: If check() hasn't been called yet
    """
    if self.results is None:
        msg = "No validation results. Call check() first."
        raise ShexValidationError(msg)

    # Handle mocked results (for testing)
    if isinstance(self.results, bool):
        return self.results

    # PyShEx returns results as a list of EvaluationResult objects
    # When validation succeeds, reason contains matching triples
    # When validation fails, reason contains error messages like
    # "Node: ... not in value set"
    # If no focus is specified, PyShEx tests all nodes;
    # we need at least one success
    if not self.results:
        return False

    # Check if at least one result succeeded (no error indicators)
    for result in self.results:
        reason = result.reason or ""
        # Common failure indicators in PyShEx error messages
        has_error = any(
            indicator in reason
            for indicator in [
                "not in value set",
                "does not match",
                "Constraint violation",
                "No matching",
                "Failed to",
            ]
        )
        if not has_error:
            return True

    return False

load_rdf()

Load RDF data from configured source.

Tries sources in order: rdf_text, rdf_file, qid (from Wikidata).

Returns:

Type	Description
ShexValidator	Self for method chaining

Raises:

Type	Description
ShexValidationError	If no valid RDF source or loading fails

Source code in gkc/shex.py

def load_rdf(self) -> "ShexValidator":
    """
    Load RDF data from configured source.

    Tries sources in order: rdf_text, rdf_file, qid (from Wikidata).

    Returns:
        Self for method chaining

    Raises:
        ShexValidationError: If no valid RDF source or loading fails
    """
    try:
        if self.rdf_text:
            self._rdf = self.rdf_text
        elif self.rdf_file:
            rdf_path = Path(self.rdf_file)
            if not rdf_path.exists():
                msg = f"RDF file not found: {self.rdf_file}"
                raise ShexValidationError(msg)
            self._rdf = rdf_path.read_text(encoding="utf-8")
        elif self.qid:
            self._rdf = fetch_entity_rdf(
                self.qid, format="ttl", user_agent=self.user_agent
            )
        else:
            raise ShexValidationError(
                "No RDF source provided. Specify qid, rdf_text, or rdf_file."
            )
    except RuntimeError as e:
        raise ShexValidationError(f"Failed to load RDF: {str(e)}") from e
    except OSError as e:
        raise ShexValidationError(f"Failed to read RDF file: {str(e)}") from e

    return self

load_specification()

Load the ShEx schema specification from configured source.

Tries sources in order: schema_text, schema_file, eid (fetch from Wikidata).

Returns:

Type	Description
ShexValidator	Self for method chaining

Raises:

Type	Description
ShexValidationError	If no valid schema source or loading fails

Source code in gkc/shex.py

def load_specification(self) -> "ShexValidator":
    """
    Load the ShEx schema specification from configured source.

    Tries sources in order: schema_text, schema_file, eid (fetch from Wikidata).

    Returns:
        Self for method chaining

    Raises:
        ShexValidationError: If no valid schema source or loading fails
    """
    try:
        if self.schema_text:
            self._schema = self.schema_text
        elif self.schema_file:
            schema_path = Path(self.schema_file)
            if not schema_path.exists():
                raise ShexValidationError(
                    f"Schema file not found: {self.schema_file}"
                )
            self._schema = schema_path.read_text(encoding="utf-8")
        elif self.eid:
            self._schema = fetch_schema_specification(self.eid, self.user_agent)
        else:
            raise ShexValidationError(
                "No schema source provided. "
                "Specify eid, schema_text, or schema_file."
            )
    except RuntimeError as e:
        raise ShexValidationError(f"Failed to load schema: {str(e)}") from e
    except OSError as e:
        msg = f"Failed to read schema file: {str(e)}"
        raise ShexValidationError(msg) from e

    return self

passes_inspection()

Check if validation passed (alias for is_valid).

Returns:

Type	Description
bool	True if validation passed, False otherwise

Raises:

Type	Description
ShexValidationError	If check() hasn't been called yet

Source code in gkc/shex.py

def passes_inspection(self) -> bool:
    """
    Check if validation passed (alias for is_valid).

    Returns:
        True if validation passed, False otherwise

    Raises:
        ShexValidationError: If check() hasn't been called yet
    """
    return self.is_valid()

Exceptions

ShexValidationError

Bases: Exception

Raised when ShEx validation encounters an error.

Source code in gkc/shex.py

class ShexValidationError(Exception):
    """Raised when ShEx validation encounters an error."""

    pass

Examples

Validating Wikidata Entities

Validate a Wikidata item against a published EntitySchema:

from gkc import ShexValidator

# Validate federally recognized tribe (Q14708404) against tribe schema (E502)
validator = ShexValidator(qid='Q14708404', eid='E502')
result = validator.check()

if result.is_valid():
    print("✓ Item conforms to EntitySchema E502")
else:
    print("✗ Item does not conform:")
    for res in result.results:
        print(f"  - {res.reason}")

Validating Local Files

Validate local RDF data against a local ShEx schema:

from gkc import ShexValidator

validator = ShexValidator(
    rdf_file='path/to/entity.ttl',
    schema_file='path/to/schema.shex'
)

result = validator.check()
print(f"Valid: {result.is_valid()}")

Mixed Sources

You can mix Wikidata and local sources:

# Wikidata entity with local schema
validator = ShexValidator(
    qid='Q42',
    schema_file='custom_schema.shex'
)
result = validator.check()

# Local RDF data with Wikidata EntitySchema
validator = ShexValidator(
    rdf_file='new_entity.ttl',
    eid='E502'
)
result = validator.check()

Using Text Strings

For programmatically generated RDF or schemas:

from gkc import ShexValidator

my_rdf = """
@prefix wd: <http://www.wikidata.org/entity/> .
@prefix wdt: <http://www.wikidata.org/prop/direct/> .

wd:Q42 wdt:P31 wd:Q5 .
"""

my_schema = """
PREFIX wdt: <http://www.wikidata.org/prop/direct/>
PREFIX wd: <http://www.wikidata.org/entity/>

<Human> {
  wdt:P31 [ wd:Q5 ]
}
"""

validator = ShexValidator(
    rdf_text=my_rdf,
    schema_text=my_schema
)

result = validator.check()

Fluent API Pattern

Load and validate step-by-step:

from gkc import ShexValidator

validator = ShexValidator(qid='Q42', eid='E502')

# Load schema
validator.load_specification()
print(f"Schema loaded: {len(validator._schema)} characters")

# Load RDF data
validator.load_rdf()
print(f"RDF loaded: {len(validator._rdf)} characters")

# Perform validation
validator.evaluate()

# Check result
if validator.passes_inspection():
    print("✓ Validation passed!")

Custom User Agent

When fetching from Wikidata, use a custom user agent:

from gkc import ShexValidator

validator = ShexValidator(
    qid='Q42',
    eid='E502',
    user_agent='MyBot/1.0 (https://example.com; [email protected])'
)

result = validator.check()

Pre-Upload Quality Check

Validate data before uploading to Wikidata:

from gkc import ShexValidator, ShexValidationError

def validate_before_upload(rdf_data: str, target_schema: str) -> bool:
    """Validate RDF data against target EntitySchema."""
    try:
        validator = ShexValidator(
            rdf_text=rdf_data,
            eid=target_schema
        )

        result = validator.check()

        if result.is_valid():
            return True
        else:
            # Log validation errors
            for res in result.results:
                print(f"Validation error: {res.reason}")
            return False

    except ShexValidationError as e:
        print(f"Validation failed: {e}")
        return False

# Use in upload workflow
if validate_before_upload(my_data, 'E502'):
    upload_to_wikidata(my_data)
else:
    print("Fix validation errors before uploading")

Batch Validation

Validate multiple entities:

from gkc import ShexValidator

entities_to_validate = [
    ('Q14708404', 'Wanapum'),
    ('Q3551781', 'Umatilla'),
    ('Q1948829', 'Muckleshoot')
]

schema_id = 'E502'  # Federally recognized tribe schema

results = {}
for qid, name in entities_to_validate:
    validator = ShexValidator(qid=qid, eid=schema_id)
    validator.check()
    results[name] = validator.is_valid()

# Report
for name, valid in results.items():
    status = "✓" if valid else "✗"
    print(f"{status} {name}")

Error Handling

Common Error Patterns

from gkc.shex import ShexValidationError, ShexValidator

try:
    validator = ShexValidator(qid='Q42', eid='E502')
    result = validator.check()

    if not result.is_valid():
        # Parse validation errors
        for res in result.results:
            reason = res.reason or ""

            if "not in value set" in reason:
                print(f"Value constraint violation: {reason}")
            elif "does not match" in reason:
                print(f"Format/type mismatch: {reason}")
            elif "Constraint violation" in reason:
                print(f"Cardinality or requirement error: {reason}")

except ShexValidationError as e:
    print(f"Validation process failed: {e}")

Handling Missing Sources

from gkc.shex import ShexValidationError, ShexValidator

try:
    validator = ShexValidator()  # No sources provided
    validator.check()
except ShexValidationError as e:
    print(f"Error: {e}")
    # Output: "No schema source provided. Specify eid, schema_text, or schema_file."

See Also

• ShEx CLI Documentation - Command-line interface
• Utilities Guide - General ShEx validation guide
• Wikidata EntitySchemas - Wikidata's schema namespace
• ShEx Primer - Learn ShEx syntax