JSON Schema

synapseclient.services.json_schema

---

JSON Schema

---

Warning

This is a beta implementation and is subject to change. Use at your own risk.

Classes

JsonSchemaVersion

Json schema version response object

ATTRIBUTE	DESCRIPTION
organization	Json Schema Organization
name	Name of the JSON schema.
semantic_version	Version of JSON schema. Defaults to None.

Source code in synapseclient/services/json_schema.py

class JsonSchemaVersion:
    """Json schema version response object

    Attributes:
        organization:     Json Schema Organization
        name:             Name of the JSON schema.
        semantic_version: Version of JSON schema. Defaults to None.
    """

    def __init__(
        self,
        organization: JsonSchemaOrganization,
        name: str,
        semantic_version: Optional[str] = None,
    ) -> None:
        self.organization = organization
        self.name = name
        self.semantic_version = semantic_version
        self.uri = None
        self.version_id = None
        self.created_on = None
        self.created_by = None
        self.json_sha256_hex = None
        self.set_service(self.organization.service)

    def __repr__(self):
        string = (
            f"JsonSchemaVersion(org={self.organization.name!r}, name={self.name!r}, "
            f"version={self.semantic_version!r})"
        )
        return string

    def set_service(self, service):
        self.service = service

    @property
    def raw(self):
        self.must_get()
        return self._raw

    def parse_response(self, response):
        self._raw = response
        self.uri = response["$id"]
        self.version_id = response["versionId"]
        self.created_on = response["createdOn"]
        self.created_by = response["createdBy"]
        self.json_sha256_hex = response["jsonSHA256Hex"]

    @classmethod
    def from_response(cls, organization, response):
        semver = response.get("semanticVersion")
        version = cls(organization, response["schemaName"], semver)
        version.parse_response(response)
        return version

    def get(self):
        """Get the JSON Schema Version"""
        if self.uri is not None:
            return True
        json_schema = self.organization.get_json_schema(self.name)
        if json_schema is None:
            return False
        raw_version = json_schema.get_version(self.semantic_version, raw=True)
        if raw_version is None:
            return False
        self.parse_response(raw_version)
        return True

    def must_get(self):
        already_exists = self.get()
        assert already_exists, (
            "This operation requires that the JSON Schema name is created first."
            "Call the 'create_version()' method to trigger the creation."
        )

    def create(
        self,
        json_schema_body: dict,
        dry_run: bool = False,
    ):
        """Create JSON schema version

        Arguments:
            json_schema_body: JSON schema body
            dry_run: Do not store to Synapse. Defaults to False.
        """
        uri = f"{self.organization.name}-{self.name}"
        if self.semantic_version:
            uri = f"{uri}-{self.semantic_version}"
        json_schema_body["$id"] = uri
        response = self.service.create_json_schema(json_schema_body, dry_run)
        if dry_run:
            return response
        raw_version = response["newVersionInfo"]
        self.parse_response(raw_version)
        return self

    def delete(self):
        """Delete the JSON schema version"""
        self.must_get()
        response = self.service.delete_json_schema(self.uri)
        return response

    @property
    def body(self):
        self.must_get()
        json_schema_body = self.service.get_json_schema_body(self.uri)
        return json_schema_body

    def expand(self):
        """Validate entities with schema"""
        self.must_get()
        response = self.service.json_schema_validation(self.uri)
        json_schema_body = response["validationSchema"]
        return json_schema_body

    def bind_to_object(self, synapse_id: str):
        """Bind schema to an entity

        Arguments:
            synapse_id: Synapse Id to bind json schema to.
        """

        self.must_get()
        response = self.service.bind_json_schema_to_entity(synapse_id, self.uri)
        return response

Functions

get

get()

Get the JSON Schema Version

Source code in synapseclient/services/json_schema.py

def get(self):
    """Get the JSON Schema Version"""
    if self.uri is not None:
        return True
    json_schema = self.organization.get_json_schema(self.name)
    if json_schema is None:
        return False
    raw_version = json_schema.get_version(self.semantic_version, raw=True)
    if raw_version is None:
        return False
    self.parse_response(raw_version)
    return True

create

create(json_schema_body: dict, dry_run: bool = False)

Create JSON schema version

PARAMETER	DESCRIPTION
json_schema_body	JSON schema body TYPE: dict
dry_run	Do not store to Synapse. Defaults to False. TYPE: bool DEFAULT: False

Source code in synapseclient/services/json_schema.py

def create(
    self,
    json_schema_body: dict,
    dry_run: bool = False,
):
    """Create JSON schema version

    Arguments:
        json_schema_body: JSON schema body
        dry_run: Do not store to Synapse. Defaults to False.
    """
    uri = f"{self.organization.name}-{self.name}"
    if self.semantic_version:
        uri = f"{uri}-{self.semantic_version}"
    json_schema_body["$id"] = uri
    response = self.service.create_json_schema(json_schema_body, dry_run)
    if dry_run:
        return response
    raw_version = response["newVersionInfo"]
    self.parse_response(raw_version)
    return self

delete

delete()

Delete the JSON schema version

Source code in synapseclient/services/json_schema.py

def delete(self):
    """Delete the JSON schema version"""
    self.must_get()
    response = self.service.delete_json_schema(self.uri)
    return response

expand

expand()

Validate entities with schema

Source code in synapseclient/services/json_schema.py

def expand(self):
    """Validate entities with schema"""
    self.must_get()
    response = self.service.json_schema_validation(self.uri)
    json_schema_body = response["validationSchema"]
    return json_schema_body

bind_to_object

bind_to_object(synapse_id: str)

Bind schema to an entity

PARAMETER	DESCRIPTION
synapse_id	Synapse Id to bind json schema to. TYPE: str

Source code in synapseclient/services/json_schema.py

def bind_to_object(self, synapse_id: str):
    """Bind schema to an entity

    Arguments:
        synapse_id: Synapse Id to bind json schema to.
    """

    self.must_get()
    response = self.service.bind_json_schema_to_entity(synapse_id, self.uri)
    return response

JsonSchema

Json schema response object

ATTRIBUTE	DESCRIPTION
organization	JSON schema organization.
name	Name of the JSON schema.

Source code in synapseclient/services/json_schema.py

class JsonSchema:
    """Json schema response object

    Attributes:
        organization:     JSON schema organization.
        name:             Name of the JSON schema.
    """

    def __init__(self, organization: JsonSchemaOrganization, name: str) -> None:
        self.organization = organization
        self.name = name
        self.id = None
        self.created_on = None
        self.created_by = None
        self._versions = dict()
        self.set_service(self.organization.service)

    def __repr__(self):
        string = f"JsonSchema(org={self.organization.name!r}, name={self.name!r})"
        return string

    def set_service(self, service):
        self.service = service

    @property
    def raw(self):
        self.must_get()
        return self._raw

    def parse_response(self, response):
        self._raw = response
        self.id = response["schemaId"]
        self.created_on = response["createdOn"]
        self.created_by = response["createdBy"]

    @classmethod
    def from_response(cls, organization, response):
        json_schema = cls(organization, response["schemaName"])
        json_schema.parse_response(response)
        return json_schema

    def get(self):
        """Get Json schema"""
        if self.id is not None:
            return True
        response = self.organization.get_json_schema(self.name, raw=True)
        if response is None:
            return False
        self.parse_response(response)
        return True

    def must_get(self):
        already_exists = self.get()
        assert already_exists, (
            "This operation requires that the JSON Schema name is created first."
            "Call the 'create_version()' method to trigger the creation."
        )

    def list_versions(self):
        """List versions of the json schema"""
        self.must_get()
        self._versions = dict()
        response = self.service.list_json_schema_versions(
            self.organization.name, self.name
        )
        for raw_version in response:
            semver = raw_version.get("semanticVersion")
            version = JsonSchemaVersion.from_response(self.organization, raw_version)
            # Handle that multiple versions can have None/null as their semver
            if semver is None:
                update_none_version = (
                    # Is this the first null version?
                    semver not in self._versions
                    # Or is the version ID higher (i.e.,  more recent)?
                    or version.version_id > self._versions[semver].version_id
                )
                if update_none_version:
                    self._versions[semver] = (raw_version, version)
            else:
                self._versions[semver] = (raw_version, version)
            # Skip versions w/o semver until the end
            if semver is not None:
                yield version
        # Return version w/o semver now (if applicable) to ensure latest is returned
        if None in self._versions:
            yield self._versions[None]

    def get_version(self, semantic_version: str = None, raw: bool = False):
        self.must_get()
        if semantic_version not in self._versions:
            list(self.list_versions())
        raw_version, version = self._versions.get(semantic_version, [None, None])
        return raw_version if raw else version

    def create(
        self,
        json_schema_body: dict,
        semantic_version: str = None,
        dry_run: bool = False,
    ):
        """Create JSON schema

        Arguments:
            json_schema_body: JSON schema body
            semantic_version: Version of JSON schema. Defaults to None.
            dry_run: Do not store to Synapse. Defaults to False.

        Returns:
            JSON schema
        """
        uri = f"{self.organization.name}-{self.name}"
        if semantic_version:
            uri = f"{uri}-{semantic_version}"
        json_schema_body["$id"] = uri
        response = self.service.create_json_schema(json_schema_body, dry_run)
        if dry_run:
            return response
        raw_version = response["newVersionInfo"]
        version = JsonSchemaVersion.from_response(self.organization, raw_version)
        self._versions[semantic_version] = (raw_version, version)
        return version

Functions

get

get()

Get Json schema

Source code in synapseclient/services/json_schema.py

def get(self):
    """Get Json schema"""
    if self.id is not None:
        return True
    response = self.organization.get_json_schema(self.name, raw=True)
    if response is None:
        return False
    self.parse_response(response)
    return True

list_versions

list_versions()

List versions of the json schema

Source code in synapseclient/services/json_schema.py

def list_versions(self):
    """List versions of the json schema"""
    self.must_get()
    self._versions = dict()
    response = self.service.list_json_schema_versions(
        self.organization.name, self.name
    )
    for raw_version in response:
        semver = raw_version.get("semanticVersion")
        version = JsonSchemaVersion.from_response(self.organization, raw_version)
        # Handle that multiple versions can have None/null as their semver
        if semver is None:
            update_none_version = (
                # Is this the first null version?
                semver not in self._versions
                # Or is the version ID higher (i.e.,  more recent)?
                or version.version_id > self._versions[semver].version_id
            )
            if update_none_version:
                self._versions[semver] = (raw_version, version)
        else:
            self._versions[semver] = (raw_version, version)
        # Skip versions w/o semver until the end
        if semver is not None:
            yield version
    # Return version w/o semver now (if applicable) to ensure latest is returned
    if None in self._versions:
        yield self._versions[None]

create

create(json_schema_body: dict, semantic_version: str = None, dry_run: bool = False)

Create JSON schema

PARAMETER	DESCRIPTION
json_schema_body	JSON schema body TYPE: dict
semantic_version	Version of JSON schema. Defaults to None. TYPE: str DEFAULT: None
dry_run	Do not store to Synapse. Defaults to False. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
	JSON schema

Source code in synapseclient/services/json_schema.py

def create(
    self,
    json_schema_body: dict,
    semantic_version: str = None,
    dry_run: bool = False,
):
    """Create JSON schema

    Arguments:
        json_schema_body: JSON schema body
        semantic_version: Version of JSON schema. Defaults to None.
        dry_run: Do not store to Synapse. Defaults to False.

    Returns:
        JSON schema
    """
    uri = f"{self.organization.name}-{self.name}"
    if semantic_version:
        uri = f"{uri}-{semantic_version}"
    json_schema_body["$id"] = uri
    response = self.service.create_json_schema(json_schema_body, dry_run)
    if dry_run:
        return response
    raw_version = response["newVersionInfo"]
    version = JsonSchemaVersion.from_response(self.organization, raw_version)
    self._versions[semantic_version] = (raw_version, version)
    return version

JsonSchemaOrganization

Json Schema Organization

ATTRIBUTE	DESCRIPTION
name	Name of JSON schema organization

Source code in synapseclient/services/json_schema.py

class JsonSchemaOrganization:
    """Json Schema Organization

    Attributes:
        name: Name of JSON schema organization
    """

    def __init__(self, name: str) -> None:
        self.name = name
        self.id = None
        self.created_on = None
        self.created_by = None
        self._json_schemas = dict()
        self._raw_json_schemas = dict()

    def __repr__(self):
        string = f"JsonSchemaOrganization(name={self.name!r})"
        return string

    def set_service(self, service):
        self.service = service

    def get(self):
        """Gets Json Schema organization"""
        if self.id is not None:
            return True
        try:
            response = self.service.get_organization(self.name)
        except SynapseHTTPError as e:
            error_msg = str(e)
            if "not found" in error_msg:
                return False
            else:
                raise e
        self.id = response["id"]
        self.created_on = response["createdOn"]
        self.created_by = response["createdBy"]
        return True

    def must_get(self):
        already_exists = self.get()
        assert already_exists, (
            "This operation requires that the organization is created first. "
            "Call the 'create()' method to trigger the creation."
        )

    @property
    def name(self):
        return self._name

    @name.setter
    def name(self, value):
        if len(value) < 6:
            raise ValueError("Name must be at least 6 characters.")
        if len(value) > 250:
            raise ValueError("Name cannot exceed 250 characters. ")
        if value[0].isdigit():
            raise ValueError("Name must not start with a number.")
        self._name = value

    @property
    def raw(self):
        self.must_get()
        return self._raw

    def parse_response(self, response):
        self._raw = response
        self.id = response["id"]
        self.created_on = response["createdOn"]
        self.created_by = response["createdBy"]

    @classmethod
    def from_response(cls, response):
        organization = cls(response["name"])
        organization.parse_response(response)
        return organization

    def create(self):
        """Create the JSON schema organization"""
        already_exists = self.get()
        if already_exists:
            return
        response = self.service.create_organization(self.name)
        self.parse_response(response)
        return self

    def delete(self):
        """Delete the JSON schema organization"""
        self.must_get()
        response = self.service.delete_organization(self.id)
        return response

    def get_acl(self):
        """Get ACL of JSON schema organization"""
        self.must_get()
        response = self.service.get_organization_acl(self.id)
        return response

    def set_acl(
        self,
        principal_ids: Sequence[int],
        access_type: Sequence[str] = DEFAULT_ACCESS,
        etag: str = None,
    ):
        """Set ACL of JSON schema organization

        Arguments:
            principal_ids: List of Synapse user or team ids.
            access_type: Access control list. Defaults to ["CHANGE_PERMISSIONS", "DELETE", "READ", "CREATE", "UPDATE"]. Defaults to DEFAULT_ACCESS.
            etag: Etag. Defaults to None.
        """
        self.must_get()
        if etag is None:
            acl = self.get_acl()
            etag = acl["etag"]
        resource_access = [
            {"principalId": principal_id, "accessType": access_type}
            for principal_id in principal_ids
        ]
        response = self.service.update_organization_acl(self.id, resource_access, etag)
        return response

    def update_acl(
        self,
        principal_ids: Sequence[int],
        access_type: Sequence[str] = DEFAULT_ACCESS,
        etag: str = None,
    ):
        """Update ACL of JSON schema organization

        Arguments:
            principal_ids: List of Synapse user or team ids.
            access_type: Access control list. Defaults to ["CHANGE_PERMISSIONS", "DELETE", "READ", "CREATE", "UPDATE"]. Defaults to DEFAULT_ACCESS.
            etag: Etag. Defaults to None.
        """
        self.must_get()
        principal_ids = set(principal_ids)
        acl = self.get_acl()
        resource_access = acl["resourceAccess"]
        if etag is None:
            etag = acl["etag"]
        for entry in resource_access:
            if entry["principalId"] in principal_ids:
                entry["accessType"] = access_type
                principal_ids.remove(entry["principalId"])
        for principal_id in principal_ids:
            entry = {
                "principalId": principal_id,
                "accessType": access_type,
            }
            resource_access.append(entry)
        response = self.service.update_organization_acl(self.id, resource_access, etag)
        return response

    def list_json_schemas(self):
        """List JSON schemas available from the organization"""
        self.must_get()
        response = self.service.list_json_schemas(self.name)
        for raw_json_schema in response:
            json_schema = JsonSchema.from_response(self, raw_json_schema)
            self._raw_json_schemas[json_schema.name] = raw_json_schema
            self._json_schemas[json_schema.name] = json_schema
            yield json_schema

    def get_json_schema(self, json_schema_name: str, raw: bool = False):
        """Get JSON schema

        Arguments:
            json_schema_name: Name of JSON schema.
            raw: Return raw JSON schema. Defaults to False.
        """
        self.must_get()
        if json_schema_name not in self._json_schemas:
            list(self.list_json_schemas())
        if raw:
            json_schema = self._raw_json_schemas.get(json_schema_name)
        else:
            json_schema = self._json_schemas.get(json_schema_name)
        return json_schema

    def create_json_schema(
        self,
        json_schema_body: dict,
        name: str = None,
        semantic_version: str = None,
        dry_run: bool = False,
    ):
        """Create JSON schema

        Arguments:
            json_schema_body: JSON schema dict
            name: Name of JSON schema. Defaults to None.
            semantic_version: Version of JSON schema. Defaults to None.
            dry_run: Don't store to Synapse. Defaults to False.
        """
        if name:
            uri = f"{self.name}-{name}"
            if semantic_version:
                uri = f"{uri}-{semantic_version}"
            json_schema_body["$id"] = uri
        else:
            assert (
                semantic_version is not None
            ), "Specify both the name and the semantic version (not just the latter)"
        response = self.service.create_json_schema(json_schema_body, dry_run)
        if dry_run:
            return response
        raw_version = response["newVersionInfo"]
        json_schema = JsonSchemaVersion.from_response(self, raw_version)
        return json_schema

Functions

get

get()

Gets Json Schema organization

Source code in synapseclient/services/json_schema.py

def get(self):
    """Gets Json Schema organization"""
    if self.id is not None:
        return True
    try:
        response = self.service.get_organization(self.name)
    except SynapseHTTPError as e:
        error_msg = str(e)
        if "not found" in error_msg:
            return False
        else:
            raise e
    self.id = response["id"]
    self.created_on = response["createdOn"]
    self.created_by = response["createdBy"]
    return True

create

create()

Create the JSON schema organization

Source code in synapseclient/services/json_schema.py

def create(self):
    """Create the JSON schema organization"""
    already_exists = self.get()
    if already_exists:
        return
    response = self.service.create_organization(self.name)
    self.parse_response(response)
    return self

delete

delete()

Delete the JSON schema organization

Source code in synapseclient/services/json_schema.py

def delete(self):
    """Delete the JSON schema organization"""
    self.must_get()
    response = self.service.delete_organization(self.id)
    return response

get_acl

get_acl()

Get ACL of JSON schema organization

Source code in synapseclient/services/json_schema.py

def get_acl(self):
    """Get ACL of JSON schema organization"""
    self.must_get()
    response = self.service.get_organization_acl(self.id)
    return response

set_acl

set_acl(principal_ids: Sequence[int], access_type: Sequence[str] = DEFAULT_ACCESS, etag: str = None)

Set ACL of JSON schema organization

PARAMETER	DESCRIPTION
principal_ids	List of Synapse user or team ids. TYPE: Sequence[int]
access_type	Access control list. Defaults to ["CHANGE_PERMISSIONS", "DELETE", "READ", "CREATE", "UPDATE"]. Defaults to DEFAULT_ACCESS. TYPE: Sequence[str] DEFAULT: DEFAULT_ACCESS
etag	Etag. Defaults to None. TYPE: str DEFAULT: None

Source code in synapseclient/services/json_schema.py

def set_acl(
    self,
    principal_ids: Sequence[int],
    access_type: Sequence[str] = DEFAULT_ACCESS,
    etag: str = None,
):
    """Set ACL of JSON schema organization

    Arguments:
        principal_ids: List of Synapse user or team ids.
        access_type: Access control list. Defaults to ["CHANGE_PERMISSIONS", "DELETE", "READ", "CREATE", "UPDATE"]. Defaults to DEFAULT_ACCESS.
        etag: Etag. Defaults to None.
    """
    self.must_get()
    if etag is None:
        acl = self.get_acl()
        etag = acl["etag"]
    resource_access = [
        {"principalId": principal_id, "accessType": access_type}
        for principal_id in principal_ids
    ]
    response = self.service.update_organization_acl(self.id, resource_access, etag)
    return response

update_acl

update_acl(principal_ids: Sequence[int], access_type: Sequence[str] = DEFAULT_ACCESS, etag: str = None)

Update ACL of JSON schema organization

PARAMETER	DESCRIPTION
principal_ids	List of Synapse user or team ids. TYPE: Sequence[int]
access_type	Access control list. Defaults to ["CHANGE_PERMISSIONS", "DELETE", "READ", "CREATE", "UPDATE"]. Defaults to DEFAULT_ACCESS. TYPE: Sequence[str] DEFAULT: DEFAULT_ACCESS
etag	Etag. Defaults to None. TYPE: str DEFAULT: None

Source code in synapseclient/services/json_schema.py

def update_acl(
    self,
    principal_ids: Sequence[int],
    access_type: Sequence[str] = DEFAULT_ACCESS,
    etag: str = None,
):
    """Update ACL of JSON schema organization

    Arguments:
        principal_ids: List of Synapse user or team ids.
        access_type: Access control list. Defaults to ["CHANGE_PERMISSIONS", "DELETE", "READ", "CREATE", "UPDATE"]. Defaults to DEFAULT_ACCESS.
        etag: Etag. Defaults to None.
    """
    self.must_get()
    principal_ids = set(principal_ids)
    acl = self.get_acl()
    resource_access = acl["resourceAccess"]
    if etag is None:
        etag = acl["etag"]
    for entry in resource_access:
        if entry["principalId"] in principal_ids:
            entry["accessType"] = access_type
            principal_ids.remove(entry["principalId"])
    for principal_id in principal_ids:
        entry = {
            "principalId": principal_id,
            "accessType": access_type,
        }
        resource_access.append(entry)
    response = self.service.update_organization_acl(self.id, resource_access, etag)
    return response

list_json_schemas

list_json_schemas()

List JSON schemas available from the organization

Source code in synapseclient/services/json_schema.py

def list_json_schemas(self):
    """List JSON schemas available from the organization"""
    self.must_get()
    response = self.service.list_json_schemas(self.name)
    for raw_json_schema in response:
        json_schema = JsonSchema.from_response(self, raw_json_schema)
        self._raw_json_schemas[json_schema.name] = raw_json_schema
        self._json_schemas[json_schema.name] = json_schema
        yield json_schema

get_json_schema

get_json_schema(json_schema_name: str, raw: bool = False)

Get JSON schema

PARAMETER	DESCRIPTION
json_schema_name	Name of JSON schema. TYPE: str
raw	Return raw JSON schema. Defaults to False. TYPE: bool DEFAULT: False

Source code in synapseclient/services/json_schema.py

def get_json_schema(self, json_schema_name: str, raw: bool = False):
    """Get JSON schema

    Arguments:
        json_schema_name: Name of JSON schema.
        raw: Return raw JSON schema. Defaults to False.
    """
    self.must_get()
    if json_schema_name not in self._json_schemas:
        list(self.list_json_schemas())
    if raw:
        json_schema = self._raw_json_schemas.get(json_schema_name)
    else:
        json_schema = self._json_schemas.get(json_schema_name)
    return json_schema

create_json_schema

create_json_schema(json_schema_body: dict, name: str = None, semantic_version: str = None, dry_run: bool = False)

Create JSON schema

PARAMETER	DESCRIPTION
json_schema_body	JSON schema dict TYPE: dict
name	Name of JSON schema. Defaults to None. TYPE: str DEFAULT: None
semantic_version	Version of JSON schema. Defaults to None. TYPE: str DEFAULT: None
dry_run	Don't store to Synapse. Defaults to False. TYPE: bool DEFAULT: False

Source code in synapseclient/services/json_schema.py

def create_json_schema(
    self,
    json_schema_body: dict,
    name: str = None,
    semantic_version: str = None,
    dry_run: bool = False,
):
    """Create JSON schema

    Arguments:
        json_schema_body: JSON schema dict
        name: Name of JSON schema. Defaults to None.
        semantic_version: Version of JSON schema. Defaults to None.
        dry_run: Don't store to Synapse. Defaults to False.
    """
    if name:
        uri = f"{self.name}-{name}"
        if semantic_version:
            uri = f"{uri}-{semantic_version}"
        json_schema_body["$id"] = uri
    else:
        assert (
            semantic_version is not None
        ), "Specify both the name and the semantic version (not just the latter)"
    response = self.service.create_json_schema(json_schema_body, dry_run)
    if dry_run:
        return response
    raw_version = response["newVersionInfo"]
    json_schema = JsonSchemaVersion.from_response(self, raw_version)
    return json_schema

JsonSchemaService

Json Schema Service

Attributes synapse: Synapse connection

Source code in synapseclient/services/json_schema.py

class JsonSchemaService:
    """Json Schema Service

    Attributes
        synapse: Synapse connection
    """

    def __init__(self, synapse: Synapse = None) -> None:
        self.synapse = synapse

    @wraps(Synapse.login)
    def login(self, *args, **kwargs):
        synapse = Synapse()
        synapse.login(*args, **kwargs)
        self.synapse = synapse

    @wraps(JsonSchemaOrganization)
    def JsonSchemaOrganization(self, *args, **kwargs):
        instance = JsonSchemaOrganization(*args, **kwargs)
        instance.set_service(self)
        return instance

    @wraps(JsonSchemaVersion)
    def JsonSchemaVersion(self, *args, **kwargs):
        instance = JsonSchemaVersion(*args, **kwargs)
        instance.set_service(self)
        return instance

    @wraps(JsonSchema)
    def JsonSchema(self, *args, **kwargs):
        instance = JsonSchema(*args, **kwargs)
        instance.set_service(self)
        return instance

    def authentication_required(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            msg = (
                f"`JsonSchemaService.{func.__name__}()` requests must be authenticated."
                " Login using the `login()` method on the existing `JsonSchemaService`"
                " instance (e.g., `js.login()` or `js.login(authToken=...)`)."
            )
            assert self.synapse is not None, msg
            try:
                result = func(self, *args, **kwargs)
            except SynapseAuthenticationError as e:
                raise SynapseAuthenticationError(msg).with_traceback(e.__traceback__)
            return result

        return wrapper

    @authentication_required
    def create_organization(self, organization_name: str):
        """Create a new organization

        Arguments:
            organization_name: JSON schema organization name
        """
        request_body = {"organizationName": organization_name}
        response = self.synapse.restPOST(
            "/schema/organization", body=json.dumps(request_body)
        )
        return response

    @authentication_required
    def get_organization(self, organization_name: str):
        """Get a organization

        Arguments:
            organization_name: JSON schema organization name
        """
        response = self.synapse.restGET(
            f"/schema/organization?name={organization_name}"
        )
        return response

    def list_organizations(self):
        """List organizations"""
        request_body = {}
        response = self.synapse._POST_paginated(
            "/schema/organization/list", request_body
        )
        return response

    @authentication_required
    def delete_organization(self, organization_id: str):
        """Delete organization

        Arguments:
            organization_id: JSON schema organization Id
        """
        response = self.synapse.restDELETE(f"/schema/organization/{organization_id}")
        return response

    @authentication_required
    def get_organization_acl(self, organization_id: str):
        """Get ACL associated with Organization

        Arguments:
            organization_id: JSON schema organization Id
        """
        response = self.synapse.restGET(f"/schema/organization/{organization_id}/acl")
        return response

    @authentication_required
    def update_organization_acl(
        self,
        organization_id: str,
        resource_access: Sequence[Mapping[str, Sequence[str]]],
        etag: str,
    ):
        """Update ACL associated with Organization

        Arguments:
            organization_id: JSON schema organization Id
            resource_access: Resource access array
            etag: Etag
        """
        request_body = {"resourceAccess": resource_access, "etag": etag}
        response = self.synapse.restPUT(
            f"/schema/organization/{organization_id}/acl", body=json.dumps(request_body)
        )
        return response

    def list_json_schemas(self, organization_name: str):
        """List JSON schemas for an organization

        Arguments:
            organization_name: JSON schema organization name
        """
        request_body = {"organizationName": organization_name}
        response = self.synapse._POST_paginated("/schema/list", request_body)
        return response

    def list_json_schema_versions(self, organization_name: str, json_schema_name: str):
        """List version information for each JSON schema

        Arguments:
            organization_name: JSON schema organization name
            json_schema_name:  JSON schema name
        """
        request_body = {
            "organizationName": organization_name,
            "schemaName": json_schema_name,
        }
        response = self.synapse._POST_paginated("/schema/version/list", request_body)
        return response

    @authentication_required
    def create_json_schema(self, json_schema_body: dict, dry_run: bool = False):
        """Create a JSON schema

        Arguments:
            json_schema_body: JSON schema body
            dry_run:          Don't store to Synapse. Default to False.
        """
        request_body = {
            "concreteType": "org.sagebionetworks.repo.model.schema.CreateSchemaRequest",
            "schema": json_schema_body,
            "dryRun": dry_run,
        }
        response = self.synapse._waitForAsync("/schema/type/create/async", request_body)
        return response

    def get_json_schema_body(self, json_schema_uri: str):
        """Get registered JSON schema with its $id

        Arguments:
            json_schema_uri: JSON schema URI
        """
        response = self.synapse.restGET(f"/schema/type/registered/{json_schema_uri}")
        return response

    @authentication_required
    def delete_json_schema(self, json_schema_uri: str):
        """Delete the given schema using its $id

        Arguments:
            json_schema_uri: JSON schema URI
        """
        response = self.synapse.restDELETE(f"/schema/type/registered/{json_schema_uri}")
        return response

    @authentication_required
    def json_schema_validation(self, json_schema_uri: str):
        """Use a JSON schema for validation

        Arguments:
            json_schema_uri: JSON schema URI
        """
        request_body = {
            "concreteType": (
                "org.sagebionetworks.repo.model.schema.GetValidationSchemaRequest"
            ),
            "$id": json_schema_uri,
        }
        response = self.synapse._waitForAsync(
            "/schema/type/validation/async", request_body
        )
        return response

    @authentication_required
    def bind_json_schema_to_entity(self, synapse_id: str, json_schema_uri: str):
        """Bind a JSON schema to an entity

        Arguments:
            synapse_id:      Synapse Id
            json_schema_uri: JSON schema URI
        """
        request_body = {"entityId": synapse_id, "schema$id": json_schema_uri}
        response = self.synapse.restPUT(
            f"/entity/{synapse_id}/schema/binding", body=json.dumps(request_body)
        )
        return response

    @authentication_required
    def get_json_schema_from_entity(self, synapse_id: str):
        """Get bound schema from entity

        Arguments:
            synapse_id:      Synapse Id
        """
        response = self.synapse.restGET(f"/entity/{synapse_id}/schema/binding")
        return response

    @authentication_required
    def delete_json_schema_from_entity(self, synapse_id: str):
        """Delete bound schema from entity

        Arguments:
            synapse_id:      Synapse Id
        """
        response = self.synapse.restDELETE(f"/entity/{synapse_id}/schema/binding")
        return response

    @authentication_required
    def validate_entity_with_json_schema(self, synapse_id: str):
        """Get validation results of an entity against bound JSON schema

        Arguments:
            synapse_id:      Synapse Id
        """
        response = self.synapse.restGET(f"/entity/{synapse_id}/schema/validation")
        return response

    @authentication_required
    def get_json_schema_validation_statistics(self, synapse_id: str):
        """Get the summary statistic of json schema validation results for
        a container entity

        Arguments:
            synapse_id:      Synapse Id
        """
        response = self.synapse.restGET(
            f"/entity/{synapse_id}/schema/validation/statistics"
        )
        return response

    @authentication_required
    def get_invalid_json_schema_validation(self, synapse_id: str):
        """Get a single page of invalid JSON schema validation results for a container Entity
        (Project or Folder).

        Arguments:
            synapse_id:      Synapse Id
        """
        request_body = {"containerId": synapse_id}
        response = self.synapse._POST_paginated(
            f"/entity/{synapse_id}/schema/validation/invalid", request_body
        )
        return response

    # The methods below are here until they are integrated with Synapse/Entity

    def bind_json_schema(self, json_schema_uri: str, entity: Union[str, Entity]):
        """Bind a JSON schema to an entity

        Arguments:
            json_schema_uri: JSON schema URI
            entity:          Synapse Entity or Synapse Id
        """
        synapse_id = id_of(entity)
        response = self.bind_json_schema_to_entity(synapse_id, json_schema_uri)
        return response

    def get_json_schema(self, entity: Union[str, Entity]):
        """Get a JSON schema associated to an Entity

        Arguments:
            entity:          Synapse Entity or Synapse Id
        """
        synapse_id = id_of(entity)
        response = self.get_json_schema_from_entity(synapse_id)
        return response

    def unbind_json_schema(self, entity: Union[str, Entity]):
        """Unbind a JSON schema from an entity

        Arguments:
            entity:          Synapse Entity or Synapse Id
        """
        synapse_id = id_of(entity)
        response = self.delete_json_schema_from_entity(synapse_id)
        return response

    def validate(self, entity: Union[str, Entity]):
        """Validate an entity based on the bound JSON schema

        Arguments:
            entity:          Synapse Entity or Synapse Id
        """
        synapse_id = id_of(entity)
        response = self.validate_entity_with_json_schema(synapse_id)
        return response

    def validation_stats(self, entity: Union[str, Entity]):
        """Get validation statistics of an entity based on the bound JSON schema

        Arguments:
            entity:          Synapse Entity or Synapse Id
        """
        synapse_id = id_of(entity)
        response = self.get_json_schema_validation_statistics(synapse_id)
        return response

    def validate_children(self, entity: Union[str, Entity]):
        """Validate an entity and it's children based on the bound JSON schema

        Arguments:
            entity:          Synapse Entity or Synapse Id of a project or folder.
        """
        synapse_id = id_of(entity)
        response = self.get_invalid_json_schema_validation(synapse_id)
        return response

Functions

create_organization

create_organization(organization_name: str)

Create a new organization

PARAMETER	DESCRIPTION
organization_name	JSON schema organization name TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def create_organization(self, organization_name: str):
    """Create a new organization

    Arguments:
        organization_name: JSON schema organization name
    """
    request_body = {"organizationName": organization_name}
    response = self.synapse.restPOST(
        "/schema/organization", body=json.dumps(request_body)
    )
    return response

get_organization

get_organization(organization_name: str)

Get a organization

PARAMETER	DESCRIPTION
organization_name	JSON schema organization name TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def get_organization(self, organization_name: str):
    """Get a organization

    Arguments:
        organization_name: JSON schema organization name
    """
    response = self.synapse.restGET(
        f"/schema/organization?name={organization_name}"
    )
    return response

list_organizations

list_organizations()

List organizations

Source code in synapseclient/services/json_schema.py

def list_organizations(self):
    """List organizations"""
    request_body = {}
    response = self.synapse._POST_paginated(
        "/schema/organization/list", request_body
    )
    return response

delete_organization

delete_organization(organization_id: str)

Delete organization

PARAMETER	DESCRIPTION
organization_id	JSON schema organization Id TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def delete_organization(self, organization_id: str):
    """Delete organization

    Arguments:
        organization_id: JSON schema organization Id
    """
    response = self.synapse.restDELETE(f"/schema/organization/{organization_id}")
    return response

get_organization_acl

get_organization_acl(organization_id: str)

Get ACL associated with Organization

PARAMETER	DESCRIPTION
organization_id	JSON schema organization Id TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def get_organization_acl(self, organization_id: str):
    """Get ACL associated with Organization

    Arguments:
        organization_id: JSON schema organization Id
    """
    response = self.synapse.restGET(f"/schema/organization/{organization_id}/acl")
    return response

update_organization_acl

update_organization_acl(organization_id: str, resource_access: Sequence[Mapping[str, Sequence[str]]], etag: str)

Update ACL associated with Organization

PARAMETER	DESCRIPTION
organization_id	JSON schema organization Id TYPE: str
resource_access	Resource access array TYPE: Sequence[Mapping[str, Sequence[str]]]
etag	Etag TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def update_organization_acl(
    self,
    organization_id: str,
    resource_access: Sequence[Mapping[str, Sequence[str]]],
    etag: str,
):
    """Update ACL associated with Organization

    Arguments:
        organization_id: JSON schema organization Id
        resource_access: Resource access array
        etag: Etag
    """
    request_body = {"resourceAccess": resource_access, "etag": etag}
    response = self.synapse.restPUT(
        f"/schema/organization/{organization_id}/acl", body=json.dumps(request_body)
    )
    return response

list_json_schemas

list_json_schemas(organization_name: str)

List JSON schemas for an organization

PARAMETER	DESCRIPTION
organization_name	JSON schema organization name TYPE: str

Source code in synapseclient/services/json_schema.py

def list_json_schemas(self, organization_name: str):
    """List JSON schemas for an organization

    Arguments:
        organization_name: JSON schema organization name
    """
    request_body = {"organizationName": organization_name}
    response = self.synapse._POST_paginated("/schema/list", request_body)
    return response

list_json_schema_versions

list_json_schema_versions(organization_name: str, json_schema_name: str)

List version information for each JSON schema

PARAMETER	DESCRIPTION
organization_name	JSON schema organization name TYPE: str
json_schema_name	JSON schema name TYPE: str

Source code in synapseclient/services/json_schema.py

def list_json_schema_versions(self, organization_name: str, json_schema_name: str):
    """List version information for each JSON schema

    Arguments:
        organization_name: JSON schema organization name
        json_schema_name:  JSON schema name
    """
    request_body = {
        "organizationName": organization_name,
        "schemaName": json_schema_name,
    }
    response = self.synapse._POST_paginated("/schema/version/list", request_body)
    return response

create_json_schema

create_json_schema(json_schema_body: dict, dry_run: bool = False)

Create a JSON schema

PARAMETER	DESCRIPTION
json_schema_body	JSON schema body TYPE: dict
dry_run	Don't store to Synapse. Default to False. TYPE: bool DEFAULT: False

Source code in synapseclient/services/json_schema.py

@authentication_required
def create_json_schema(self, json_schema_body: dict, dry_run: bool = False):
    """Create a JSON schema

    Arguments:
        json_schema_body: JSON schema body
        dry_run:          Don't store to Synapse. Default to False.
    """
    request_body = {
        "concreteType": "org.sagebionetworks.repo.model.schema.CreateSchemaRequest",
        "schema": json_schema_body,
        "dryRun": dry_run,
    }
    response = self.synapse._waitForAsync("/schema/type/create/async", request_body)
    return response

get_json_schema_body

get_json_schema_body(json_schema_uri: str)

Get registered JSON schema with its $id

PARAMETER	DESCRIPTION
json_schema_uri	JSON schema URI TYPE: str

Source code in synapseclient/services/json_schema.py

def get_json_schema_body(self, json_schema_uri: str):
    """Get registered JSON schema with its $id

    Arguments:
        json_schema_uri: JSON schema URI
    """
    response = self.synapse.restGET(f"/schema/type/registered/{json_schema_uri}")
    return response

delete_json_schema

delete_json_schema(json_schema_uri: str)

Delete the given schema using its $id

PARAMETER	DESCRIPTION
json_schema_uri	JSON schema URI TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def delete_json_schema(self, json_schema_uri: str):
    """Delete the given schema using its $id

    Arguments:
        json_schema_uri: JSON schema URI
    """
    response = self.synapse.restDELETE(f"/schema/type/registered/{json_schema_uri}")
    return response

json_schema_validation

json_schema_validation(json_schema_uri: str)

Use a JSON schema for validation

PARAMETER	DESCRIPTION
json_schema_uri	JSON schema URI TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def json_schema_validation(self, json_schema_uri: str):
    """Use a JSON schema for validation

    Arguments:
        json_schema_uri: JSON schema URI
    """
    request_body = {
        "concreteType": (
            "org.sagebionetworks.repo.model.schema.GetValidationSchemaRequest"
        ),
        "$id": json_schema_uri,
    }
    response = self.synapse._waitForAsync(
        "/schema/type/validation/async", request_body
    )
    return response

bind_json_schema_to_entity

bind_json_schema_to_entity(synapse_id: str, json_schema_uri: str)

Bind a JSON schema to an entity

PARAMETER	DESCRIPTION
synapse_id	Synapse Id TYPE: str
json_schema_uri	JSON schema URI TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def bind_json_schema_to_entity(self, synapse_id: str, json_schema_uri: str):
    """Bind a JSON schema to an entity

    Arguments:
        synapse_id:      Synapse Id
        json_schema_uri: JSON schema URI
    """
    request_body = {"entityId": synapse_id, "schema$id": json_schema_uri}
    response = self.synapse.restPUT(
        f"/entity/{synapse_id}/schema/binding", body=json.dumps(request_body)
    )
    return response

get_json_schema_from_entity

get_json_schema_from_entity(synapse_id: str)

Get bound schema from entity

PARAMETER	DESCRIPTION
synapse_id	Synapse Id TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def get_json_schema_from_entity(self, synapse_id: str):
    """Get bound schema from entity

    Arguments:
        synapse_id:      Synapse Id
    """
    response = self.synapse.restGET(f"/entity/{synapse_id}/schema/binding")
    return response

delete_json_schema_from_entity

delete_json_schema_from_entity(synapse_id: str)

Delete bound schema from entity

PARAMETER	DESCRIPTION
synapse_id	Synapse Id TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def delete_json_schema_from_entity(self, synapse_id: str):
    """Delete bound schema from entity

    Arguments:
        synapse_id:      Synapse Id
    """
    response = self.synapse.restDELETE(f"/entity/{synapse_id}/schema/binding")
    return response

validate_entity_with_json_schema

validate_entity_with_json_schema(synapse_id: str)

Get validation results of an entity against bound JSON schema

PARAMETER	DESCRIPTION
synapse_id	Synapse Id TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def validate_entity_with_json_schema(self, synapse_id: str):
    """Get validation results of an entity against bound JSON schema

    Arguments:
        synapse_id:      Synapse Id
    """
    response = self.synapse.restGET(f"/entity/{synapse_id}/schema/validation")
    return response

get_json_schema_validation_statistics

get_json_schema_validation_statistics(synapse_id: str)

Get the summary statistic of json schema validation results for a container entity

PARAMETER	DESCRIPTION
synapse_id	Synapse Id TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def get_json_schema_validation_statistics(self, synapse_id: str):
    """Get the summary statistic of json schema validation results for
    a container entity

    Arguments:
        synapse_id:      Synapse Id
    """
    response = self.synapse.restGET(
        f"/entity/{synapse_id}/schema/validation/statistics"
    )
    return response

get_invalid_json_schema_validation

get_invalid_json_schema_validation(synapse_id: str)

Get a single page of invalid JSON schema validation results for a container Entity (Project or Folder).

PARAMETER	DESCRIPTION
synapse_id	Synapse Id TYPE: str

Source code in synapseclient/services/json_schema.py

@authentication_required
def get_invalid_json_schema_validation(self, synapse_id: str):
    """Get a single page of invalid JSON schema validation results for a container Entity
    (Project or Folder).

    Arguments:
        synapse_id:      Synapse Id
    """
    request_body = {"containerId": synapse_id}
    response = self.synapse._POST_paginated(
        f"/entity/{synapse_id}/schema/validation/invalid", request_body
    )
    return response

bind_json_schema

bind_json_schema(json_schema_uri: str, entity: Union[str, Entity])

Bind a JSON schema to an entity

PARAMETER	DESCRIPTION
json_schema_uri	JSON schema URI TYPE: str
entity	Synapse Entity or Synapse Id TYPE: Union[str, Entity]

Source code in synapseclient/services/json_schema.py

def bind_json_schema(self, json_schema_uri: str, entity: Union[str, Entity]):
    """Bind a JSON schema to an entity

    Arguments:
        json_schema_uri: JSON schema URI
        entity:          Synapse Entity or Synapse Id
    """
    synapse_id = id_of(entity)
    response = self.bind_json_schema_to_entity(synapse_id, json_schema_uri)
    return response

get_json_schema

get_json_schema(entity: Union[str, Entity])

Get a JSON schema associated to an Entity

PARAMETER	DESCRIPTION
entity	Synapse Entity or Synapse Id TYPE: Union[str, Entity]

Source code in synapseclient/services/json_schema.py

def get_json_schema(self, entity: Union[str, Entity]):
    """Get a JSON schema associated to an Entity

    Arguments:
        entity:          Synapse Entity or Synapse Id
    """
    synapse_id = id_of(entity)
    response = self.get_json_schema_from_entity(synapse_id)
    return response

unbind_json_schema

unbind_json_schema(entity: Union[str, Entity])

Unbind a JSON schema from an entity

PARAMETER	DESCRIPTION
entity	Synapse Entity or Synapse Id TYPE: Union[str, Entity]

Source code in synapseclient/services/json_schema.py

def unbind_json_schema(self, entity: Union[str, Entity]):
    """Unbind a JSON schema from an entity

    Arguments:
        entity:          Synapse Entity or Synapse Id
    """
    synapse_id = id_of(entity)
    response = self.delete_json_schema_from_entity(synapse_id)
    return response

validate

validate(entity: Union[str, Entity])

Validate an entity based on the bound JSON schema

PARAMETER	DESCRIPTION
entity	Synapse Entity or Synapse Id TYPE: Union[str, Entity]

Source code in synapseclient/services/json_schema.py

def validate(self, entity: Union[str, Entity]):
    """Validate an entity based on the bound JSON schema

    Arguments:
        entity:          Synapse Entity or Synapse Id
    """
    synapse_id = id_of(entity)
    response = self.validate_entity_with_json_schema(synapse_id)
    return response

validation_stats

validation_stats(entity: Union[str, Entity])

Get validation statistics of an entity based on the bound JSON schema

PARAMETER	DESCRIPTION
entity	Synapse Entity or Synapse Id TYPE: Union[str, Entity]

Source code in synapseclient/services/json_schema.py

def validation_stats(self, entity: Union[str, Entity]):
    """Get validation statistics of an entity based on the bound JSON schema

    Arguments:
        entity:          Synapse Entity or Synapse Id
    """
    synapse_id = id_of(entity)
    response = self.get_json_schema_validation_statistics(synapse_id)
    return response

validate_children

validate_children(entity: Union[str, Entity])

Validate an entity and it's children based on the bound JSON schema

PARAMETER	DESCRIPTION
entity	Synapse Entity or Synapse Id of a project or folder. TYPE: Union[str, Entity]

Source code in synapseclient/services/json_schema.py

def validate_children(self, entity: Union[str, Entity]):
    """Validate an entity and it's children based on the bound JSON schema

    Arguments:
        entity:          Synapse Entity or Synapse Id of a project or folder.
    """
    synapse_id = id_of(entity)
    response = self.get_invalid_json_schema_validation(synapse_id)
    return response

Functions