ContainerEntityJSONSchema

synapseclient.models.mixins.ContainerEntityJSONSchema

Bases: ContainerEntityJSONSchemaProtocol, BaseJSONSchema

Mixin class to provide JSON schema functionality. This class is intended to be used with classes that represent Synapse entities. It provides methods to bind, delete, and validate JSON schemas associated with the entity.

Source code in synapseclient/models/mixins/json_schema.py

@async_to_sync
class ContainerEntityJSONSchema(ContainerEntityJSONSchemaProtocol, BaseJSONSchema):
    """
    Mixin class to provide JSON schema functionality.
    This class is intended to be used with classes that represent Synapse entities.
    It provides methods to bind, delete, and validate JSON schemas associated with the entity.
    """

    async def get_schema_validation_statistics_async(
        self, *, synapse_client: Optional["Synapse"] = None
    ) -> JSONSchemaValidationStatistics:
        """
        Get validation statistics for a container entity.

        Arguments:
            synapse_client (Optional[Synapse], optional): The Synapse client instance. If not provided,
                the last created instance from the Synapse class constructor will be used.

        Returns:
            The validation statistics.

        Example: Using this function
            Retrieving validation statistics for a folder. This example demonstrates
            how to get validation statistics for a container entity after creating
            entities with various validation states. Set the `PROJECT_NAME` variable to your project name.

            ```python
            import asyncio
            import time
            from synapseclient import Synapse
            from synapseclient.models import File, Folder

            syn = Synapse()
            syn.login()

            # Define Project and JSON schema info
            PROJECT_NAME = "test_json_schema_project"  # replace with your project name
            FILE_PATH = "~/Sample.txt"  # replace with your test file path

            PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
            ORG_NAME = "UniqueOrg"  # replace with your organization name
            SCHEMA_NAME = "myTestSchema"  # replace with your schema name
            FOLDER_NAME = "test_script_folder"
            VERSION = "0.0.1"
            SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

            # Create organization (if not already created)
            js = syn.service("json_schema")
            all_orgs = js.list_organizations()
            for org in all_orgs:
                if org["name"] == ORG_NAME:
                    print(f"Organization {ORG_NAME} already exists: {org}")
                    break
            else:
                print(f"Creating organization {ORG_NAME}.")
                created_organization = js.create_organization(ORG_NAME)
                print(f"Created organization: {created_organization}")

            my_test_org = js.JsonSchemaOrganization(ORG_NAME)
            test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

            if not test_schema:
                # Create the schema (if not already created)
                schema_definition = {
                    "$id": "mySchema",
                    "type": "object",
                    "properties": {
                        "foo": {"type": "string"},
                        "baz": {"type": "string", "const": "example_value"},
                        "bar": {"type": "integer"},
                    },
                    "required": ["foo"]
                }
                test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
                print(f"Created new schema: {SCHEMA_NAME}")


            async def main():
                # Create a test folder
                test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
                await test_folder.store_async()
                print(f"Created test folder: {FOLDER_NAME}")

                # Bind JSON schema to the folder
                bound_schema = await test_folder.bind_schema_async(
                    json_schema_uri=SCHEMA_URI,
                    enable_derived_annotations=True
                )
                print(f"Bound schema to folder: {bound_schema}")

                # Create files within the folder with invalid annotations to generate statistics
                invalid_file1 = File(
                    path=FILE_PATH,  # assumes you have something here or adjust path
                    parent_id=test_folder.id
                )
                invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
                await invalid_file1.store_async()
                print("Created file with invalid annotations")
                time.sleep(2)  # Allow time for processing

                stats = await test_folder.get_schema_validation_statistics_async()
                print(f"Validation statistics: {stats}")

            asyncio.run(main())
            ```
        """
        response = await get_json_schema_validation_statistics(
            synapse_id=self.id, synapse_client=synapse_client
        )
        return JSONSchemaValidationStatistics(
            container_id=response.get("containerId", ""),
            total_number_of_children=response.get("totalNumberOfChildren", None),
            number_of_valid_children=response.get("numberOfValidChildren", None),
            number_of_invalid_children=response.get("numberOfInvalidChildren", None),
            number_of_unknown_children=response.get("numberOfUnknownChildren", None),
        )

    @skip_async_to_sync
    async def get_invalid_validation_async(
        self, *, synapse_client: Optional["Synapse"] = None
    ) -> AsyncGenerator[InvalidJSONSchemaValidation, None]:
        """
        Get invalid JSON schema validation results for a container entity.

        Arguments:
            synapse_client (Optional[Synapse], optional): The Synapse client instance. If not provided,
                the last created instance from the Synapse class constructor will be used.

        Yields:
            An object containing the validation response, all validation messages,
                                         and the validation exception details.

        Example: Using this function
            Retrieving invalid validation results for a folder. This example demonstrates
            how to get detailed invalid validation results for child entities within a container.
            Set the `PROJECT_NAME` variable to your project name.

            ```python
            import asyncio
            import time
            from synapseclient import Synapse
            from synapseclient.models import File, Folder

            syn = Synapse()
            syn.login()

            # Define Project and JSON schema info
            PROJECT_NAME = "test_json_schema_project"  # replace with your project name
            FILE_PATH = "~/Sample.txt"  # replace with your test file path

            PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
            ORG_NAME = "UniqueOrg"  # replace with your organization name
            SCHEMA_NAME = "myTestSchema"  # replace with your schema name
            FOLDER_NAME = "test_script_folder"
            VERSION = "0.0.1"
            SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

            # Create organization (if not already created)
            js = syn.service("json_schema")
            all_orgs = js.list_organizations()
            for org in all_orgs:
                if org["name"] == ORG_NAME:
                    print(f"Organization {ORG_NAME} already exists: {org}")
                    break
            else:
                print(f"Creating organization {ORG_NAME}.")
                created_organization = js.create_organization(ORG_NAME)
                print(f"Created organization: {created_organization}")

            my_test_org = js.JsonSchemaOrganization(ORG_NAME)
            test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

            if not test_schema:
                # Create the schema (if not already created)
                schema_definition = {
                    "$id": "mySchema",
                    "type": "object",
                    "properties": {
                        "foo": {"type": "string"},
                        "baz": {"type": "string", "const": "example_value"},
                        "bar": {"type": "integer"},
                    },
                    "required": ["foo"]
                }
                test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
                print(f"Created new schema: {SCHEMA_NAME}")


            async def main():
                # Create a test folder
                test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
                await test_folder.store_async()
                print(f"Created test folder: {FOLDER_NAME}")

                # Bind JSON schema to the folder
                bound_schema = await test_folder.bind_schema_async(
                    json_schema_uri=SCHEMA_URI,
                    enable_derived_annotations=True
                )
                print(f"Bound schema to folder: {bound_schema}")

                # Create files within the folder with invalid annotations to generate invalid results
                invalid_file1 = File(
                    path=FILE_PATH,  # assumes you have something here or adjust path
                    parent_id=test_folder.id
                )
                invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
                await invalid_file1.store_async()
                print("Created file with invalid annotations")
                time.sleep(2)  # Allow time for processing

                print("Getting invalid validation results...")
                gen = test_folder.get_invalid_validation_async(synapse_client=syn)
                async for child in gen:
                    print(f"Invalid validation item: {child}")

            asyncio.run(main())
            ```
        """
        gen = get_invalid_json_schema_validation(
            synapse_client=synapse_client, synapse_id=self.id
        )
        async for item in gen:
            yield InvalidJSONSchemaValidation(
                validation_response=JSONSchemaValidation(
                    object_id=item.get("objectId", None),
                    object_type=item.get("objectType", None),
                    object_etag=item.get("objectEtag", None),
                    id=item.get("schema$id", None),
                    is_valid=item.get("isValid", None),
                    validated_on=item.get("validatedOn", None),
                ),
                validation_error_message=item.get("validationErrorMessage", None),
                all_validation_messages=item.get("allValidationMessages", []),
                validation_exception=ValidationException(
                    pointer_to_violation=item.get("validationException", {}).get(
                        "pointerToViolation", None
                    ),
                    message=item.get("validationException", {}).get("message", None),
                    schema_location=item.get("validationException", {}).get(
                        "schemaLocation", None
                    ),
                    causing_exceptions=[
                        CausingException(
                            keyword=ce.get("keyword", None),
                            pointer_to_violation=ce.get("pointerToViolation", None),
                            message=ce.get("message", None),
                            schema_location=ce.get("schemaLocation", None),
                            causing_exceptions=[
                                CausingException(
                                    keyword=nce.get("keyword", None),
                                    pointer_to_violation=nce.get(
                                        "pointerToViolation", None
                                    ),
                                    message=nce.get("message", None),
                                    schema_location=nce.get("schemaLocation", None),
                                )
                                for nce in ce.get("causingExceptions", [])
                            ],
                        )
                        for ce in item.get("validationException", {}).get(
                            "causingExceptions", []
                        )
                    ],
                ),
            )

    def get_invalid_validation(
        self, *, synapse_client: Optional["Synapse"] = None
    ) -> Generator[InvalidJSONSchemaValidation, None, None]:
        """
        Get invalid JSON schema validation results for a container entity.

        Arguments:
            synapse_client (Optional[Synapse], optional): The Synapse client instance. If not provided,
                the last created instance from the Synapse class constructor will be used.

        Yields:
            An object containing the validation response, all validation messages,
                                         and the validation exception details.

        Example: Using this function
            Retrieving invalid validation results for a folder. This example demonstrates
            how to get detailed invalid validation results for child entities within a container.
            Set the `PROJECT_NAME` variable to your project name.

            ```python
            import time
            from synapseclient import Synapse
            from synapseclient.models import File, Folder

            syn = Synapse()
            syn.login()

            # Define Project and JSON schema info
            PROJECT_NAME = "test_json_schema_project"  # replace with your project name
            FILE_PATH = "~/Sample.txt"  # replace with your test file path

            PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
            ORG_NAME = "UniqueOrg"  # replace with your organization name
            SCHEMA_NAME = "myTestSchema"  # replace with your schema name
            FOLDER_NAME = "test_script_folder"
            VERSION = "0.0.1"
            SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

            # Create organization (if not already created)
            js = syn.service("json_schema")
            all_orgs = js.list_organizations()
            for org in all_orgs:
                if org["name"] == ORG_NAME:
                    print(f"Organization {ORG_NAME} already exists: {org}")
                    break
            else:
                print(f"Creating organization {ORG_NAME}.")
                created_organization = js.create_organization(ORG_NAME)
                print(f"Created organization: {created_organization}")

            my_test_org = js.JsonSchemaOrganization(ORG_NAME)
            test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

            if not test_schema:
                # Create the schema (if not already created)
                schema_definition = {
                    "$id": "mySchema",
                    "type": "object",
                    "properties": {
                        "foo": {"type": "string"},
                        "baz": {"type": "string", "const": "example_value"},
                        "bar": {"type": "integer"},
                    },
                    "required": ["foo"]
                }
                test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
                print(f"Created new schema: {SCHEMA_NAME}")

            # Create a test folder
            test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
            test_folder.store()
            print(f"Created test folder: {FOLDER_NAME}")

            # Bind JSON schema to the folder
            bound_schema = test_folder.bind_schema(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to folder: {bound_schema}")

            # Create files within the folder with invalid annotations to generate invalid results
            invalid_file1 = File(
                path=FILE_PATH,  # assumes you have something here or adjust path
                parent_id=test_folder.id
            )
            invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
            invalid_file1.store()
            print("Created file with invalid annotations")
            time.sleep(2)  # Allow time for processing

            # Get invalid validation results
            invalid_results = test_folder.get_invalid_validation(synapse_client=syn)
            print(f"Invalid validation items found:")
            for child in invalid_results:
                print(f"Invalid validation item: {child}")
            ```
        """
        gen = get_invalid_json_schema_validation_sync(
            synapse_client=synapse_client, synapse_id=self.id
        )

        for item in gen:
            yield InvalidJSONSchemaValidation(
                validation_response=JSONSchemaValidation(
                    object_id=item.get("objectId", None),
                    object_type=item.get("objectType", None),
                    object_etag=item.get("objectEtag", None),
                    id=item.get("schema$id", None),
                    is_valid=item.get("isValid", None),
                    validated_on=item.get("validatedOn", None),
                ),
                validation_error_message=item.get("validationErrorMessage", None),
                all_validation_messages=item.get("allValidationMessages", []),
                validation_exception=ValidationException(
                    pointer_to_violation=item.get("validationException", {}).get(
                        "pointerToViolation", None
                    ),
                    message=item.get("validationException", {}).get("message", None),
                    schema_location=item.get("validationException", {}).get(
                        "schemaLocation", None
                    ),
                    causing_exceptions=[
                        CausingException(
                            keyword=ce.get("keyword", None),
                            pointer_to_violation=ce.get("pointerToViolation", None),
                            message=ce.get("message", None),
                            schema_location=ce.get("schemaLocation", None),
                            causing_exceptions=[
                                CausingException(
                                    keyword=nce.get("keyword", None),
                                    pointer_to_violation=nce.get(
                                        "pointerToViolation", None
                                    ),
                                    message=nce.get("message", None),
                                    schema_location=nce.get("schemaLocation", None),
                                )
                                for nce in ce.get("causingExceptions", [])
                            ],
                        )
                        for ce in item.get("validationException", {}).get(
                            "causingExceptions", []
                        )
                    ],
                ),
            )

Functions

get_schema_validation_statistics_async async

get_schema_validation_statistics_async(*, synapse_client: Optional[Synapse] = None) -> JSONSchemaValidationStatistics

Get validation statistics for a container entity.

PARAMETER	DESCRIPTION
synapse_client	The Synapse client instance. If not provided, the last created instance from the Synapse class constructor will be used. TYPE: Optional[Synapse] DEFAULT: None

RETURNS	DESCRIPTION
JSONSchemaValidationStatistics	The validation statistics.

Using this function

Retrieving validation statistics for a folder. This example demonstrates how to get validation statistics for a container entity after creating entities with various validation states. Set the PROJECT_NAME variable to your project name.

import asyncio
import time
from synapseclient import Synapse
from synapseclient.models import File, Folder

syn = Synapse()
syn.login()

# Define Project and JSON schema info
PROJECT_NAME = "test_json_schema_project"  # replace with your project name
FILE_PATH = "~/Sample.txt"  # replace with your test file path

PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
ORG_NAME = "UniqueOrg"  # replace with your organization name
SCHEMA_NAME = "myTestSchema"  # replace with your schema name
FOLDER_NAME = "test_script_folder"
VERSION = "0.0.1"
SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

# Create organization (if not already created)
js = syn.service("json_schema")
all_orgs = js.list_organizations()
for org in all_orgs:
    if org["name"] == ORG_NAME:
        print(f"Organization {ORG_NAME} already exists: {org}")
        break
else:
    print(f"Creating organization {ORG_NAME}.")
    created_organization = js.create_organization(ORG_NAME)
    print(f"Created organization: {created_organization}")

my_test_org = js.JsonSchemaOrganization(ORG_NAME)
test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

if not test_schema:
    # Create the schema (if not already created)
    schema_definition = {
        "$id": "mySchema",
        "type": "object",
        "properties": {
            "foo": {"type": "string"},
            "baz": {"type": "string", "const": "example_value"},
            "bar": {"type": "integer"},
        },
        "required": ["foo"]
    }
    test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
    print(f"Created new schema: {SCHEMA_NAME}")


async def main():
    # Create a test folder
    test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
    await test_folder.store_async()
    print(f"Created test folder: {FOLDER_NAME}")

    # Bind JSON schema to the folder
    bound_schema = await test_folder.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to folder: {bound_schema}")

    # Create files within the folder with invalid annotations to generate statistics
    invalid_file1 = File(
        path=FILE_PATH,  # assumes you have something here or adjust path
        parent_id=test_folder.id
    )
    invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
    await invalid_file1.store_async()
    print("Created file with invalid annotations")
    time.sleep(2)  # Allow time for processing

    stats = await test_folder.get_schema_validation_statistics_async()
    print(f"Validation statistics: {stats}")

asyncio.run(main())

Source code in synapseclient/models/mixins/json_schema.py

async def get_schema_validation_statistics_async(
    self, *, synapse_client: Optional["Synapse"] = None
) -> JSONSchemaValidationStatistics:
    """
    Get validation statistics for a container entity.

    Arguments:
        synapse_client (Optional[Synapse], optional): The Synapse client instance. If not provided,
            the last created instance from the Synapse class constructor will be used.

    Returns:
        The validation statistics.

    Example: Using this function
        Retrieving validation statistics for a folder. This example demonstrates
        how to get validation statistics for a container entity after creating
        entities with various validation states. Set the `PROJECT_NAME` variable to your project name.

        ```python
        import asyncio
        import time
        from synapseclient import Synapse
        from synapseclient.models import File, Folder

        syn = Synapse()
        syn.login()

        # Define Project and JSON schema info
        PROJECT_NAME = "test_json_schema_project"  # replace with your project name
        FILE_PATH = "~/Sample.txt"  # replace with your test file path

        PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
        ORG_NAME = "UniqueOrg"  # replace with your organization name
        SCHEMA_NAME = "myTestSchema"  # replace with your schema name
        FOLDER_NAME = "test_script_folder"
        VERSION = "0.0.1"
        SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

        # Create organization (if not already created)
        js = syn.service("json_schema")
        all_orgs = js.list_organizations()
        for org in all_orgs:
            if org["name"] == ORG_NAME:
                print(f"Organization {ORG_NAME} already exists: {org}")
                break
        else:
            print(f"Creating organization {ORG_NAME}.")
            created_organization = js.create_organization(ORG_NAME)
            print(f"Created organization: {created_organization}")

        my_test_org = js.JsonSchemaOrganization(ORG_NAME)
        test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

        if not test_schema:
            # Create the schema (if not already created)
            schema_definition = {
                "$id": "mySchema",
                "type": "object",
                "properties": {
                    "foo": {"type": "string"},
                    "baz": {"type": "string", "const": "example_value"},
                    "bar": {"type": "integer"},
                },
                "required": ["foo"]
            }
            test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
            print(f"Created new schema: {SCHEMA_NAME}")


        async def main():
            # Create a test folder
            test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
            await test_folder.store_async()
            print(f"Created test folder: {FOLDER_NAME}")

            # Bind JSON schema to the folder
            bound_schema = await test_folder.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to folder: {bound_schema}")

            # Create files within the folder with invalid annotations to generate statistics
            invalid_file1 = File(
                path=FILE_PATH,  # assumes you have something here or adjust path
                parent_id=test_folder.id
            )
            invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
            await invalid_file1.store_async()
            print("Created file with invalid annotations")
            time.sleep(2)  # Allow time for processing

            stats = await test_folder.get_schema_validation_statistics_async()
            print(f"Validation statistics: {stats}")

        asyncio.run(main())
        ```
    """
    response = await get_json_schema_validation_statistics(
        synapse_id=self.id, synapse_client=synapse_client
    )
    return JSONSchemaValidationStatistics(
        container_id=response.get("containerId", ""),
        total_number_of_children=response.get("totalNumberOfChildren", None),
        number_of_valid_children=response.get("numberOfValidChildren", None),
        number_of_invalid_children=response.get("numberOfInvalidChildren", None),
        number_of_unknown_children=response.get("numberOfUnknownChildren", None),
    )

get_invalid_validation_async async

get_invalid_validation_async(*, synapse_client: Optional[Synapse] = None) -> AsyncGenerator[InvalidJSONSchemaValidation, None]

Get invalid JSON schema validation results for a container entity.

PARAMETER	DESCRIPTION
synapse_client	The Synapse client instance. If not provided, the last created instance from the Synapse class constructor will be used. TYPE: Optional[Synapse] DEFAULT: None

YIELDS	DESCRIPTION
AsyncGenerator[InvalidJSONSchemaValidation, None]	An object containing the validation response, all validation messages, and the validation exception details.

Using this function

Retrieving invalid validation results for a folder. This example demonstrates how to get detailed invalid validation results for child entities within a container. Set the PROJECT_NAME variable to your project name.

import asyncio
import time
from synapseclient import Synapse
from synapseclient.models import File, Folder

syn = Synapse()
syn.login()

# Define Project and JSON schema info
PROJECT_NAME = "test_json_schema_project"  # replace with your project name
FILE_PATH = "~/Sample.txt"  # replace with your test file path

PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
ORG_NAME = "UniqueOrg"  # replace with your organization name
SCHEMA_NAME = "myTestSchema"  # replace with your schema name
FOLDER_NAME = "test_script_folder"
VERSION = "0.0.1"
SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

# Create organization (if not already created)
js = syn.service("json_schema")
all_orgs = js.list_organizations()
for org in all_orgs:
    if org["name"] == ORG_NAME:
        print(f"Organization {ORG_NAME} already exists: {org}")
        break
else:
    print(f"Creating organization {ORG_NAME}.")
    created_organization = js.create_organization(ORG_NAME)
    print(f"Created organization: {created_organization}")

my_test_org = js.JsonSchemaOrganization(ORG_NAME)
test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

if not test_schema:
    # Create the schema (if not already created)
    schema_definition = {
        "$id": "mySchema",
        "type": "object",
        "properties": {
            "foo": {"type": "string"},
            "baz": {"type": "string", "const": "example_value"},
            "bar": {"type": "integer"},
        },
        "required": ["foo"]
    }
    test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
    print(f"Created new schema: {SCHEMA_NAME}")


async def main():
    # Create a test folder
    test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
    await test_folder.store_async()
    print(f"Created test folder: {FOLDER_NAME}")

    # Bind JSON schema to the folder
    bound_schema = await test_folder.bind_schema_async(
        json_schema_uri=SCHEMA_URI,
        enable_derived_annotations=True
    )
    print(f"Bound schema to folder: {bound_schema}")

    # Create files within the folder with invalid annotations to generate invalid results
    invalid_file1 = File(
        path=FILE_PATH,  # assumes you have something here or adjust path
        parent_id=test_folder.id
    )
    invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
    await invalid_file1.store_async()
    print("Created file with invalid annotations")
    time.sleep(2)  # Allow time for processing

    print("Getting invalid validation results...")
    gen = test_folder.get_invalid_validation_async(synapse_client=syn)
    async for child in gen:
        print(f"Invalid validation item: {child}")

asyncio.run(main())

Source code in synapseclient/models/mixins/json_schema.py

@skip_async_to_sync
async def get_invalid_validation_async(
    self, *, synapse_client: Optional["Synapse"] = None
) -> AsyncGenerator[InvalidJSONSchemaValidation, None]:
    """
    Get invalid JSON schema validation results for a container entity.

    Arguments:
        synapse_client (Optional[Synapse], optional): The Synapse client instance. If not provided,
            the last created instance from the Synapse class constructor will be used.

    Yields:
        An object containing the validation response, all validation messages,
                                     and the validation exception details.

    Example: Using this function
        Retrieving invalid validation results for a folder. This example demonstrates
        how to get detailed invalid validation results for child entities within a container.
        Set the `PROJECT_NAME` variable to your project name.

        ```python
        import asyncio
        import time
        from synapseclient import Synapse
        from synapseclient.models import File, Folder

        syn = Synapse()
        syn.login()

        # Define Project and JSON schema info
        PROJECT_NAME = "test_json_schema_project"  # replace with your project name
        FILE_PATH = "~/Sample.txt"  # replace with your test file path

        PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
        ORG_NAME = "UniqueOrg"  # replace with your organization name
        SCHEMA_NAME = "myTestSchema"  # replace with your schema name
        FOLDER_NAME = "test_script_folder"
        VERSION = "0.0.1"
        SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

        # Create organization (if not already created)
        js = syn.service("json_schema")
        all_orgs = js.list_organizations()
        for org in all_orgs:
            if org["name"] == ORG_NAME:
                print(f"Organization {ORG_NAME} already exists: {org}")
                break
        else:
            print(f"Creating organization {ORG_NAME}.")
            created_organization = js.create_organization(ORG_NAME)
            print(f"Created organization: {created_organization}")

        my_test_org = js.JsonSchemaOrganization(ORG_NAME)
        test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

        if not test_schema:
            # Create the schema (if not already created)
            schema_definition = {
                "$id": "mySchema",
                "type": "object",
                "properties": {
                    "foo": {"type": "string"},
                    "baz": {"type": "string", "const": "example_value"},
                    "bar": {"type": "integer"},
                },
                "required": ["foo"]
            }
            test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
            print(f"Created new schema: {SCHEMA_NAME}")


        async def main():
            # Create a test folder
            test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
            await test_folder.store_async()
            print(f"Created test folder: {FOLDER_NAME}")

            # Bind JSON schema to the folder
            bound_schema = await test_folder.bind_schema_async(
                json_schema_uri=SCHEMA_URI,
                enable_derived_annotations=True
            )
            print(f"Bound schema to folder: {bound_schema}")

            # Create files within the folder with invalid annotations to generate invalid results
            invalid_file1 = File(
                path=FILE_PATH,  # assumes you have something here or adjust path
                parent_id=test_folder.id
            )
            invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
            await invalid_file1.store_async()
            print("Created file with invalid annotations")
            time.sleep(2)  # Allow time for processing

            print("Getting invalid validation results...")
            gen = test_folder.get_invalid_validation_async(synapse_client=syn)
            async for child in gen:
                print(f"Invalid validation item: {child}")

        asyncio.run(main())
        ```
    """
    gen = get_invalid_json_schema_validation(
        synapse_client=synapse_client, synapse_id=self.id
    )
    async for item in gen:
        yield InvalidJSONSchemaValidation(
            validation_response=JSONSchemaValidation(
                object_id=item.get("objectId", None),
                object_type=item.get("objectType", None),
                object_etag=item.get("objectEtag", None),
                id=item.get("schema$id", None),
                is_valid=item.get("isValid", None),
                validated_on=item.get("validatedOn", None),
            ),
            validation_error_message=item.get("validationErrorMessage", None),
            all_validation_messages=item.get("allValidationMessages", []),
            validation_exception=ValidationException(
                pointer_to_violation=item.get("validationException", {}).get(
                    "pointerToViolation", None
                ),
                message=item.get("validationException", {}).get("message", None),
                schema_location=item.get("validationException", {}).get(
                    "schemaLocation", None
                ),
                causing_exceptions=[
                    CausingException(
                        keyword=ce.get("keyword", None),
                        pointer_to_violation=ce.get("pointerToViolation", None),
                        message=ce.get("message", None),
                        schema_location=ce.get("schemaLocation", None),
                        causing_exceptions=[
                            CausingException(
                                keyword=nce.get("keyword", None),
                                pointer_to_violation=nce.get(
                                    "pointerToViolation", None
                                ),
                                message=nce.get("message", None),
                                schema_location=nce.get("schemaLocation", None),
                            )
                            for nce in ce.get("causingExceptions", [])
                        ],
                    )
                    for ce in item.get("validationException", {}).get(
                        "causingExceptions", []
                    )
                ],
            ),
        )

get_invalid_validation

get_invalid_validation(*, synapse_client: Optional[Synapse] = None) -> Generator[InvalidJSONSchemaValidation, None, None]

Get invalid JSON schema validation results for a container entity.

PARAMETER	DESCRIPTION
synapse_client	The Synapse client instance. If not provided, the last created instance from the Synapse class constructor will be used. TYPE: Optional[Synapse] DEFAULT: None

YIELDS	DESCRIPTION
InvalidJSONSchemaValidation	An object containing the validation response, all validation messages, and the validation exception details.

Using this function

Retrieving invalid validation results for a folder. This example demonstrates how to get detailed invalid validation results for child entities within a container. Set the PROJECT_NAME variable to your project name.

import time
from synapseclient import Synapse
from synapseclient.models import File, Folder

syn = Synapse()
syn.login()

# Define Project and JSON schema info
PROJECT_NAME = "test_json_schema_project"  # replace with your project name
FILE_PATH = "~/Sample.txt"  # replace with your test file path

PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
ORG_NAME = "UniqueOrg"  # replace with your organization name
SCHEMA_NAME = "myTestSchema"  # replace with your schema name
FOLDER_NAME = "test_script_folder"
VERSION = "0.0.1"
SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

# Create organization (if not already created)
js = syn.service("json_schema")
all_orgs = js.list_organizations()
for org in all_orgs:
    if org["name"] == ORG_NAME:
        print(f"Organization {ORG_NAME} already exists: {org}")
        break
else:
    print(f"Creating organization {ORG_NAME}.")
    created_organization = js.create_organization(ORG_NAME)
    print(f"Created organization: {created_organization}")

my_test_org = js.JsonSchemaOrganization(ORG_NAME)
test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

if not test_schema:
    # Create the schema (if not already created)
    schema_definition = {
        "$id": "mySchema",
        "type": "object",
        "properties": {
            "foo": {"type": "string"},
            "baz": {"type": "string", "const": "example_value"},
            "bar": {"type": "integer"},
        },
        "required": ["foo"]
    }
    test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
    print(f"Created new schema: {SCHEMA_NAME}")

# Create a test folder
test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
test_folder.store()
print(f"Created test folder: {FOLDER_NAME}")

# Bind JSON schema to the folder
bound_schema = test_folder.bind_schema(
    json_schema_uri=SCHEMA_URI,
    enable_derived_annotations=True
)
print(f"Bound schema to folder: {bound_schema}")

# Create files within the folder with invalid annotations to generate invalid results
invalid_file1 = File(
    path=FILE_PATH,  # assumes you have something here or adjust path
    parent_id=test_folder.id
)
invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
invalid_file1.store()
print("Created file with invalid annotations")
time.sleep(2)  # Allow time for processing

# Get invalid validation results
invalid_results = test_folder.get_invalid_validation(synapse_client=syn)
print(f"Invalid validation items found:")
for child in invalid_results:
    print(f"Invalid validation item: {child}")

Source code in synapseclient/models/mixins/json_schema.py

def get_invalid_validation(
    self, *, synapse_client: Optional["Synapse"] = None
) -> Generator[InvalidJSONSchemaValidation, None, None]:
    """
    Get invalid JSON schema validation results for a container entity.

    Arguments:
        synapse_client (Optional[Synapse], optional): The Synapse client instance. If not provided,
            the last created instance from the Synapse class constructor will be used.

    Yields:
        An object containing the validation response, all validation messages,
                                     and the validation exception details.

    Example: Using this function
        Retrieving invalid validation results for a folder. This example demonstrates
        how to get detailed invalid validation results for child entities within a container.
        Set the `PROJECT_NAME` variable to your project name.

        ```python
        import time
        from synapseclient import Synapse
        from synapseclient.models import File, Folder

        syn = Synapse()
        syn.login()

        # Define Project and JSON schema info
        PROJECT_NAME = "test_json_schema_project"  # replace with your project name
        FILE_PATH = "~/Sample.txt"  # replace with your test file path

        PROJECT_ID = syn.findEntityId(name=PROJECT_NAME)
        ORG_NAME = "UniqueOrg"  # replace with your organization name
        SCHEMA_NAME = "myTestSchema"  # replace with your schema name
        FOLDER_NAME = "test_script_folder"
        VERSION = "0.0.1"
        SCHEMA_URI = f"{ORG_NAME}-{SCHEMA_NAME}-{VERSION}"

        # Create organization (if not already created)
        js = syn.service("json_schema")
        all_orgs = js.list_organizations()
        for org in all_orgs:
            if org["name"] == ORG_NAME:
                print(f"Organization {ORG_NAME} already exists: {org}")
                break
        else:
            print(f"Creating organization {ORG_NAME}.")
            created_organization = js.create_organization(ORG_NAME)
            print(f"Created organization: {created_organization}")

        my_test_org = js.JsonSchemaOrganization(ORG_NAME)
        test_schema = my_test_org.get_json_schema(SCHEMA_NAME)

        if not test_schema:
            # Create the schema (if not already created)
            schema_definition = {
                "$id": "mySchema",
                "type": "object",
                "properties": {
                    "foo": {"type": "string"},
                    "baz": {"type": "string", "const": "example_value"},
                    "bar": {"type": "integer"},
                },
                "required": ["foo"]
            }
            test_schema = my_test_org.create_json_schema(schema_definition, SCHEMA_NAME, VERSION)
            print(f"Created new schema: {SCHEMA_NAME}")

        # Create a test folder
        test_folder = Folder(name=FOLDER_NAME, parent_id=PROJECT_ID)
        test_folder.store()
        print(f"Created test folder: {FOLDER_NAME}")

        # Bind JSON schema to the folder
        bound_schema = test_folder.bind_schema(
            json_schema_uri=SCHEMA_URI,
            enable_derived_annotations=True
        )
        print(f"Bound schema to folder: {bound_schema}")

        # Create files within the folder with invalid annotations to generate invalid results
        invalid_file1 = File(
            path=FILE_PATH,  # assumes you have something here or adjust path
            parent_id=test_folder.id
        )
        invalid_file1.annotations = {"foo": 123, "bar": "not_an_integer"}  # both invalid
        invalid_file1.store()
        print("Created file with invalid annotations")
        time.sleep(2)  # Allow time for processing

        # Get invalid validation results
        invalid_results = test_folder.get_invalid_validation(synapse_client=syn)
        print(f"Invalid validation items found:")
        for child in invalid_results:
            print(f"Invalid validation item: {child}")
        ```
    """
    gen = get_invalid_json_schema_validation_sync(
        synapse_client=synapse_client, synapse_id=self.id
    )

    for item in gen:
        yield InvalidJSONSchemaValidation(
            validation_response=JSONSchemaValidation(
                object_id=item.get("objectId", None),
                object_type=item.get("objectType", None),
                object_etag=item.get("objectEtag", None),
                id=item.get("schema$id", None),
                is_valid=item.get("isValid", None),
                validated_on=item.get("validatedOn", None),
            ),
            validation_error_message=item.get("validationErrorMessage", None),
            all_validation_messages=item.get("allValidationMessages", []),
            validation_exception=ValidationException(
                pointer_to_violation=item.get("validationException", {}).get(
                    "pointerToViolation", None
                ),
                message=item.get("validationException", {}).get("message", None),
                schema_location=item.get("validationException", {}).get(
                    "schemaLocation", None
                ),
                causing_exceptions=[
                    CausingException(
                        keyword=ce.get("keyword", None),
                        pointer_to_violation=ce.get("pointerToViolation", None),
                        message=ce.get("message", None),
                        schema_location=ce.get("schemaLocation", None),
                        causing_exceptions=[
                            CausingException(
                                keyword=nce.get("keyword", None),
                                pointer_to_violation=nce.get(
                                    "pointerToViolation", None
                                ),
                                message=nce.get("message", None),
                                schema_location=nce.get("schemaLocation", None),
                            )
                            for nce in ce.get("causingExceptions", [])
                        ],
                    )
                    for ce in item.get("validationException", {}).get(
                        "causingExceptions", []
                    )
                ],
            ),
        )

synapseclient.models.mixins.JSONSchemaValidationStatistics dataclass

Represents the summary statistics of JSON schema validation results for a container.

Source code in synapseclient/models/mixins/json_schema.py

@dataclass
class JSONSchemaValidationStatistics:
    """Represents the summary statistics of JSON schema validation results for a container."""

    container_id: str
    """The Synapse ID of the parent container."""

    total_number_of_children: int
    """The total number of child entities."""

    number_of_valid_children: int
    """The number of children that passed validation."""

    number_of_invalid_children: int
    """The number of children that failed validation."""

    number_of_unknown_children: int
    """The number of children with unknown validation status."""

Attributes

container_id instance-attribute

container_id: str

The Synapse ID of the parent container.

total_number_of_children instance-attribute

total_number_of_children: int

The total number of child entities.

number_of_valid_children instance-attribute

number_of_valid_children: int

The number of children that passed validation.

number_of_invalid_children instance-attribute

number_of_invalid_children: int

The number of children that failed validation.

number_of_unknown_children instance-attribute

number_of_unknown_children: int

The number of children with unknown validation status.

synapseclient.models.mixins.InvalidJSONSchemaValidation dataclass

Represents the response for invalid JSON schema validation results.

Source code in synapseclient/models/mixins/json_schema.py

@dataclass
class InvalidJSONSchemaValidation:
    """Represents the response for invalid JSON schema validation results."""

    validation_response: JSONSchemaValidation
    """The validation response object."""

    validation_error_message: str
    """A message describing the validation error."""

    all_validation_messages: List[str]
    """A list of all validation messages."""

    validation_exception: ValidationException
    """The validation exception details."""

Attributes

validation_response instance-attribute

validation_response: JSONSchemaValidation

The validation response object.

validation_error_message instance-attribute

validation_error_message: str

A message describing the validation error.

all_validation_messages instance-attribute

all_validation_messages: List[str]

A list of all validation messages.

validation_exception instance-attribute

validation_exception: ValidationException

The validation exception details.