Skip to content

The JsonDataModel class

JsonDataModel

A backend-independent dynamic data model.

This structure is the one flowing in the pipelines as the backend data models are only used for the variable/data model declaration.

Parameters:

Name Type Description Default
schema dict

The JSON object's schema. If not provided, uses the data_model to infer it.

None
value dict

The JSON object's value. If not provided, uses the data_model to infer it.

None
data_model DataModel | JsonDataModel

The data_model to use to infer the schema and value.

None
name str

Optional. The name of the data model, automatically inferred if not provided.

None

Examples:

Creating a JsonDataModel with a DataModel's schema and value:

class Query(synalinks.DataModel):
    query: str = synalinks.Field(
        description="The user query",
    )

value = {"query": "What is the capital of France?"}

data_model = JsonDataModel(
    schema=Query.schema(),
    value=value,
)

Creating a JsonDataModel with a data_model:

class Query(synalinks.DataModel):
    query: str = synalinks.Field(
        description="The user query",
    )

query_instance = Query(
    query="What is the capital of France?"
)
data_model = JsonDataModel(
    data_model=query_instance,
)

Creating a JsonDataModel with to_json_data_model():

class Query(synalinks.DataModel):
    query: str = synalinks.Field(
        description="The user query",
    )

data_model = Query(
    query="What is the capital of France?",
).to_json_data_model()
Source code in synalinks/src/backend/common/json_data_model.py
@synalinks_export("synalinks.JsonDataModel")
class JsonDataModel:
    """A backend-independent dynamic data model.

    This structure is the one flowing in the pipelines as
    the backend data models are only used for the variable/data model declaration.

    Args:
        schema (dict): The JSON object's schema. If not provided,
            uses the data_model to infer it.
        value (dict): The JSON object's value. If not provided,
            uses the data_model to infer it.
        data_model (DataModel | JsonDataModel): The data_model to use to
            infer the schema and value.
        name (str): Optional. The name of the data model, automatically
            inferred if not provided.

    Examples:

    **Creating a `JsonDataModel` with a DataModel's schema and value:**

    ```python
    class Query(synalinks.DataModel):
        query: str = synalinks.Field(
            description="The user query",
        )

    value = {"query": "What is the capital of France?"}

    data_model = JsonDataModel(
        schema=Query.schema(),
        value=value,
    )
    ```

    **Creating a `JsonDataModel` with a data_model:**

    ```python
    class Query(synalinks.DataModel):
        query: str = synalinks.Field(
            description="The user query",
        )

    query_instance = Query(
        query="What is the capital of France?"
    )
    data_model = JsonDataModel(
        data_model=query_instance,
    )
    ```

    **Creating a `JsonDataModel` with `to_json_data_model()`:**

    ```python
    class Query(synalinks.DataModel):
        query: str = synalinks.Field(
            description="The user query",
        )

    data_model = Query(
        query="What is the capital of France?",
    ).to_json_data_model()
    ```
    """

    def __init__(
        self,
        schema=None,
        value=None,
        data_model=None,
        name=None,
    ):
        name = name or auto_name(self.__class__.__name__)
        self._name = name
        self._schema = None
        self._value = None

        if not data_model and not schema and not value:
            raise ValueError("Initializing without arguments is not permited.")
        if not schema and not data_model:
            raise ValueError(
                "You should specify at least one argument between data_model or schema"
            )
        if not value and not data_model:
            raise ValueError(
                "You should specify at least one argument between data_model or value"
            )
        if data_model:
            if not schema:
                schema = data_model.schema()
            if not value:
                if inspect.isclass(data_model):
                    raise ValueError(
                        "Couldn't get the JSON data from the data_model, "
                        "the data_model needs to be instanciated."
                    )
                value = data_model.json()

        self._schema = standardize_schema(schema)
        self._value = value

    def to_symbolic_data_model(self):
        """Converts the JsonDataModel to a SymbolicDataModel.

        Returns:
            (SymbolicDataModel): The symbolic data model.
        """
        return SymbolicDataModel(schema=self._schema)

    def json(self):
        """Alias for the JSON object's value.

        Returns:
            (dict): The current value of the JSON object.
        """
        return self.value()

    def value(self):
        """The current value of the JSON object.

        Returns:
            (dict): The current value of the JSON object.
        """
        return self._value

    def schema(self):
        """Gets the schema of the JSON object.

        Returns:
            (dict): The JSON schema.
        """
        return self._schema

    def pretty_schema(self):
        """Get a pretty version of the JSON schema for display.

        Returns:
            (dict): The indented JSON schema.
        """
        return json.dumps(self.schema(), indent=2)

    def pretty_json(self):
        """Get a pretty version of the JSON object for display.

        Returns:
            (str): The indented JSON object.
        """
        return json.dumps(self.json(), indent=2)

    @property
    def name(self):
        """The name of the Json object."""
        return self._name

    def __add__(self, other):
        """Concatenates this data model with another.

        Args:
            other (JsonDataModel | DataModel):
                The other data model to concatenate with.

        Returns:
            (JsonDataModel): The concatenated data model.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.Concat().call(self, other),
        )

    def __radd__(self, other):
        """Concatenates another data model with this one.

        Args:
            other (JsonDataModel | DataModel):
                The other data model to concatenate with.

        Returns:
            (JsonDataModel): The concatenated data model.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.Concat().call(other, self),
        )

    def __and__(self, other):
        """Perform a `logical_and` with another data model.

        If one of them is None, output None. If both are provided,
        then concatenates this data model with the other.

        Args:
            other (JsonDataModel | DataModel): The other data model to concatenate with.

        Returns:
            (JsonDataModel | None): The concatenated data model or None
                based on the `logical_and` table.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.And().call(self, other),
        )

    def __rand__(self, other):
        """Perform a `logical_and` (reverse) with another data model.

        If one of them is None, output None. If both are provided,
        then concatenates the other data model with this one.

        Args:
            other (JsonDataModel | DataModel): The other data model to concatenate with.

        Returns:
            (JsonDataModel | None): The concatenated data model or None
                based on the `logical_and` table.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.And().call(other, self),
        )

    def __or__(self, other):
        """Perform a `logical_or` with another data model

        If one of them is None, output the other one. If both are provided,
        then concatenates this data model with the other.

        Args:
            other (JsonDataModel | DataModel): The other data model to concatenate with.

        Returns:
            (JsonDataModel | None): The concatenation of data model if both are provided,
                or the non-None data model or None if none are provided.
                (See `logical_or` table).
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.Or().call(self, other),
        )

    def __ror__(self, other):
        """Perform a `logical_or` (reverse) with another data model

        If one of them is None, output the other one. If both are provided,
        then concatenates the other data model with this one.

        Args:
            other (JsonDataModel | DataModel): The other data model to concatenate with.

        Returns:
            (JsonDataModel | None): The concatenation of data model if both are provided,
                or the non-None data model or None if none are provided.
                (See `logical_or` table).
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.Or().call(other, self),
        )

    def factorize(self):
        """Factorizes the data model.

        Returns:
            (JsonDataModel): The factorized data model.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.Factorize().call(self),
        )

    def in_mask(self, mask=None, recursive=True):
        """Applies a mask to **keep only** specified keys of the data model.

        Args:
            mask (list): The mask to be applied.
            recursive (bool): Optional. Whether to apply the mask recursively.
                Defaults to True.

        Returns:
            (JsonDataModel): The data model with the mask applied.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.InMask(mask=mask, recursive=recursive).call(self),
        )

    def out_mask(self, mask=None, recursive=True):
        """Applies a mask to **remove** specified keys of the data model.

        Args:
            mask (list): The mask to be applied.
            recursive (bool): Optional. Whether to apply the mask recursively.
                Defaults to True.

        Returns:
            (JsonDataModel): The data model with the mask applied.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.OutMask(mask=mask, recursive=recursive).call(self),
        )

    def prefix(self, prefix=None):
        """Add a prefix to **all** the data model fields (non-recursive).

        Args:
            prefix (str): the prefix to add.

        Returns:
            (JsonDataModel): The data model with the prefix added.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.Prefix(prefix=prefix).call(self),
        )

    def suffix(self, suffix=None):
        """Add a suffix to **all** the data model fields (non-recursive).

        Args:
            suffix (str): the suffix to add.

        Returns:
            (JsonDataModel): The data model with the suffix added.
        """
        from synalinks.src import ops

        return asyncio.get_event_loop().run_until_complete(
            ops.Suffix(suffix=suffix).call(self),
        )

    def get(self, key):
        """Get wrapper to make easier to access fields.

        Args:
            key (str): The key to access.
        """
        return self.json().get(key)

    def update(self, kv_dict):
        """Update wrapper to make easier to modify fields.

        Args:
            kv_dict (dict): The key/value dict to update.
        """
        self.json().update(kv_dict)

    def __repr__(self):
        return f"<JsonDataModel schema={self._schema}, value={self._value}>"

name property

The name of the Json object.

__add__(other)

Concatenates this data model with another.

Parameters:

Name Type Description Default
other JsonDataModel | DataModel

The other data model to concatenate with.

required

Returns:

Type Description
JsonDataModel

The concatenated data model.

Source code in synalinks/src/backend/common/json_data_model.py
def __add__(self, other):
    """Concatenates this data model with another.

    Args:
        other (JsonDataModel | DataModel):
            The other data model to concatenate with.

    Returns:
        (JsonDataModel): The concatenated data model.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.Concat().call(self, other),
    )

__and__(other)

Perform a logical_and with another data model.

If one of them is None, output None. If both are provided, then concatenates this data model with the other.

Parameters:

Name Type Description Default
other JsonDataModel | DataModel

The other data model to concatenate with.

required

Returns:

Type Description
JsonDataModel | None

The concatenated data model or None based on the logical_and table.

Source code in synalinks/src/backend/common/json_data_model.py
def __and__(self, other):
    """Perform a `logical_and` with another data model.

    If one of them is None, output None. If both are provided,
    then concatenates this data model with the other.

    Args:
        other (JsonDataModel | DataModel): The other data model to concatenate with.

    Returns:
        (JsonDataModel | None): The concatenated data model or None
            based on the `logical_and` table.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.And().call(self, other),
    )

__or__(other)

Perform a logical_or with another data model

If one of them is None, output the other one. If both are provided, then concatenates this data model with the other.

Parameters:

Name Type Description Default
other JsonDataModel | DataModel

The other data model to concatenate with.

required

Returns:

Type Description
JsonDataModel | None

The concatenation of data model if both are provided, or the non-None data model or None if none are provided. (See logical_or table).

Source code in synalinks/src/backend/common/json_data_model.py
def __or__(self, other):
    """Perform a `logical_or` with another data model

    If one of them is None, output the other one. If both are provided,
    then concatenates this data model with the other.

    Args:
        other (JsonDataModel | DataModel): The other data model to concatenate with.

    Returns:
        (JsonDataModel | None): The concatenation of data model if both are provided,
            or the non-None data model or None if none are provided.
            (See `logical_or` table).
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.Or().call(self, other),
    )

__radd__(other)

Concatenates another data model with this one.

Parameters:

Name Type Description Default
other JsonDataModel | DataModel

The other data model to concatenate with.

required

Returns:

Type Description
JsonDataModel

The concatenated data model.

Source code in synalinks/src/backend/common/json_data_model.py
def __radd__(self, other):
    """Concatenates another data model with this one.

    Args:
        other (JsonDataModel | DataModel):
            The other data model to concatenate with.

    Returns:
        (JsonDataModel): The concatenated data model.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.Concat().call(other, self),
    )

__rand__(other)

Perform a logical_and (reverse) with another data model.

If one of them is None, output None. If both are provided, then concatenates the other data model with this one.

Parameters:

Name Type Description Default
other JsonDataModel | DataModel

The other data model to concatenate with.

required

Returns:

Type Description
JsonDataModel | None

The concatenated data model or None based on the logical_and table.

Source code in synalinks/src/backend/common/json_data_model.py
def __rand__(self, other):
    """Perform a `logical_and` (reverse) with another data model.

    If one of them is None, output None. If both are provided,
    then concatenates the other data model with this one.

    Args:
        other (JsonDataModel | DataModel): The other data model to concatenate with.

    Returns:
        (JsonDataModel | None): The concatenated data model or None
            based on the `logical_and` table.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.And().call(other, self),
    )

__ror__(other)

Perform a logical_or (reverse) with another data model

If one of them is None, output the other one. If both are provided, then concatenates the other data model with this one.

Parameters:

Name Type Description Default
other JsonDataModel | DataModel

The other data model to concatenate with.

required

Returns:

Type Description
JsonDataModel | None

The concatenation of data model if both are provided, or the non-None data model or None if none are provided. (See logical_or table).

Source code in synalinks/src/backend/common/json_data_model.py
def __ror__(self, other):
    """Perform a `logical_or` (reverse) with another data model

    If one of them is None, output the other one. If both are provided,
    then concatenates the other data model with this one.

    Args:
        other (JsonDataModel | DataModel): The other data model to concatenate with.

    Returns:
        (JsonDataModel | None): The concatenation of data model if both are provided,
            or the non-None data model or None if none are provided.
            (See `logical_or` table).
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.Or().call(other, self),
    )

factorize()

Factorizes the data model.

Returns:

Type Description
JsonDataModel

The factorized data model.

Source code in synalinks/src/backend/common/json_data_model.py
def factorize(self):
    """Factorizes the data model.

    Returns:
        (JsonDataModel): The factorized data model.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.Factorize().call(self),
    )

get(key)

Get wrapper to make easier to access fields.

Parameters:

Name Type Description Default
key str

The key to access.

required
Source code in synalinks/src/backend/common/json_data_model.py
def get(self, key):
    """Get wrapper to make easier to access fields.

    Args:
        key (str): The key to access.
    """
    return self.json().get(key)

in_mask(mask=None, recursive=True)

Applies a mask to keep only specified keys of the data model.

Parameters:

Name Type Description Default
mask list

The mask to be applied.

None
recursive bool

Optional. Whether to apply the mask recursively. Defaults to True.

True

Returns:

Type Description
JsonDataModel

The data model with the mask applied.

Source code in synalinks/src/backend/common/json_data_model.py
def in_mask(self, mask=None, recursive=True):
    """Applies a mask to **keep only** specified keys of the data model.

    Args:
        mask (list): The mask to be applied.
        recursive (bool): Optional. Whether to apply the mask recursively.
            Defaults to True.

    Returns:
        (JsonDataModel): The data model with the mask applied.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.InMask(mask=mask, recursive=recursive).call(self),
    )

json()

Alias for the JSON object's value.

Returns:

Type Description
dict

The current value of the JSON object.

Source code in synalinks/src/backend/common/json_data_model.py
def json(self):
    """Alias for the JSON object's value.

    Returns:
        (dict): The current value of the JSON object.
    """
    return self.value()

out_mask(mask=None, recursive=True)

Applies a mask to remove specified keys of the data model.

Parameters:

Name Type Description Default
mask list

The mask to be applied.

None
recursive bool

Optional. Whether to apply the mask recursively. Defaults to True.

True

Returns:

Type Description
JsonDataModel

The data model with the mask applied.

Source code in synalinks/src/backend/common/json_data_model.py
def out_mask(self, mask=None, recursive=True):
    """Applies a mask to **remove** specified keys of the data model.

    Args:
        mask (list): The mask to be applied.
        recursive (bool): Optional. Whether to apply the mask recursively.
            Defaults to True.

    Returns:
        (JsonDataModel): The data model with the mask applied.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.OutMask(mask=mask, recursive=recursive).call(self),
    )

prefix(prefix=None)

Add a prefix to all the data model fields (non-recursive).

Parameters:

Name Type Description Default
prefix str

the prefix to add.

None

Returns:

Type Description
JsonDataModel

The data model with the prefix added.

Source code in synalinks/src/backend/common/json_data_model.py
def prefix(self, prefix=None):
    """Add a prefix to **all** the data model fields (non-recursive).

    Args:
        prefix (str): the prefix to add.

    Returns:
        (JsonDataModel): The data model with the prefix added.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.Prefix(prefix=prefix).call(self),
    )

pretty_json()

Get a pretty version of the JSON object for display.

Returns:

Type Description
str

The indented JSON object.

Source code in synalinks/src/backend/common/json_data_model.py
def pretty_json(self):
    """Get a pretty version of the JSON object for display.

    Returns:
        (str): The indented JSON object.
    """
    return json.dumps(self.json(), indent=2)

pretty_schema()

Get a pretty version of the JSON schema for display.

Returns:

Type Description
dict

The indented JSON schema.

Source code in synalinks/src/backend/common/json_data_model.py
def pretty_schema(self):
    """Get a pretty version of the JSON schema for display.

    Returns:
        (dict): The indented JSON schema.
    """
    return json.dumps(self.schema(), indent=2)

schema()

Gets the schema of the JSON object.

Returns:

Type Description
dict

The JSON schema.

Source code in synalinks/src/backend/common/json_data_model.py
def schema(self):
    """Gets the schema of the JSON object.

    Returns:
        (dict): The JSON schema.
    """
    return self._schema

suffix(suffix=None)

Add a suffix to all the data model fields (non-recursive).

Parameters:

Name Type Description Default
suffix str

the suffix to add.

None

Returns:

Type Description
JsonDataModel

The data model with the suffix added.

Source code in synalinks/src/backend/common/json_data_model.py
def suffix(self, suffix=None):
    """Add a suffix to **all** the data model fields (non-recursive).

    Args:
        suffix (str): the suffix to add.

    Returns:
        (JsonDataModel): The data model with the suffix added.
    """
    from synalinks.src import ops

    return asyncio.get_event_loop().run_until_complete(
        ops.Suffix(suffix=suffix).call(self),
    )

to_symbolic_data_model()

Converts the JsonDataModel to a SymbolicDataModel.

Returns:

Type Description
SymbolicDataModel

The symbolic data model.

Source code in synalinks/src/backend/common/json_data_model.py
def to_symbolic_data_model(self):
    """Converts the JsonDataModel to a SymbolicDataModel.

    Returns:
        (SymbolicDataModel): The symbolic data model.
    """
    return SymbolicDataModel(schema=self._schema)

update(kv_dict)

Update wrapper to make easier to modify fields.

Parameters:

Name Type Description Default
kv_dict dict

The key/value dict to update.

required
Source code in synalinks/src/backend/common/json_data_model.py
def update(self, kv_dict):
    """Update wrapper to make easier to modify fields.

    Args:
        kv_dict (dict): The key/value dict to update.
    """
    self.json().update(kv_dict)

value()

The current value of the JSON object.

Returns:

Type Description
dict

The current value of the JSON object.

Source code in synalinks/src/backend/common/json_data_model.py
def value(self):
    """The current value of the JSON object.

    Returns:
        (dict): The current value of the JSON object.
    """
    return self._value

is_json_data_model(x)

Returns whether x is a backend-independent data model.

Parameters:

Name Type Description Default
x any

The object to check.

required

Returns:

Type Description
bool

True if x is a backend-independent data model, False otherwise.

Source code in synalinks/src/backend/common/json_data_model.py
@synalinks_export(
    [
        "synalinks.utils.is_json_data_model",
        "synalinks.backend.is_json_data_model",
    ]
)
def is_json_data_model(x):
    """Returns whether `x` is a backend-independent data model.

    Args:
        x (any): The object to check.

    Returns:
        (bool): True if `x` is a backend-independent data model, False otherwise.
    """
    return isinstance(x, JsonDataModel)