JSON Ops

And

Bases: Operation

Perform a logical And operation between data models.

Source code in synalinks/src/ops/json.py

class And(Operation):
    """Perform a logical `And` operation between data models."""

    def __init__(
        self,
        name=None,
        description=None,
    ):
        super().__init__(
            name=name,
            description=description,
        )

    async def call(self, x1, x2):
        if x1 and x2:
            value = concatenate_json(x1.json(), x2.json())
            schema = concatenate_schema(x1.schema(), x2.schema())
            return JsonDataModel(value=value, schema=schema, name=self.name)
        elif x1 and not x2:
            return None
        elif not x1 and x2:
            return None
        else:
            return None

    async def compute_output_spec(self, x1, x2):
        schema = concatenate_schema(x1.schema(), x2.schema())
        return SymbolicDataModel(schema=schema, name=self.name)

Concat

Bases: Operation

Concatenate two data models together.

Source code in synalinks/src/ops/json.py

class Concat(Operation):
    """Concatenate two data models together."""

    def __init__(
        self,
        name=None,
        description=None,
    ):
        super().__init__(
            name=name,
            description=description,
        )

    async def call(self, x1, x2):
        if not x1:
            raise ValueError(f"Received x1={x1} and x2={x2}")
        if not x2:
            raise ValueError(f"Received x1={x1} and x2={x2}")
        value = concatenate_json(x1.json(), x2.json())
        schema = concatenate_schema(x1.schema(), x2.schema())
        return JsonDataModel(value=value, schema=schema, name=self.name)

    async def compute_output_spec(self, x1, x2):
        schema = concatenate_schema(x1.schema(), x2.schema())
        return SymbolicDataModel(schema=schema, name=self.name)

Factorize

Bases: Operation

Factorize a data model by grouping similar properties together.

Source code in synalinks/src/ops/json.py

class Factorize(Operation):
    """Factorize a data model by grouping similar properties together."""

    def __init__(
        self,
        name=None,
        description=None,
    ):
        super().__init__(
            name=name,
            description=description,
        )

    async def call(self, x):
        value = factorize_json(x.json())
        schema = factorize_schema(x.schema())
        return JsonDataModel(value=value, schema=schema, name=self.name)

    async def compute_output_spec(self, x):
        schema = factorize_schema(x.schema())
        return SymbolicDataModel(schema=schema, name=self.name)

InMask

Bases: Operation

Keep specific fields of a data model.

Source code in synalinks/src/ops/json.py

class InMask(Operation):
    """Keep specific fields of a data model."""

    def __init__(
        self,
        mask=None,
        recursive=True,
        name=None,
        description=None,
    ):
        super().__init__(
            name=name,
            description=description,
        )
        self.mask = mask
        self.recursive = recursive

    async def call(self, x):
        value = in_mask_json(x.json(), mask=self.mask, recursive=self.recursive)
        schema = in_mask_schema(x.schema(), mask=self.mask, recursive=self.recursive)
        return JsonDataModel(value=value, schema=schema, name=self.name)

    async def compute_output_spec(self, x):
        schema = in_mask_schema(x.schema(), mask=self.mask, recursive=self.recursive)
        return SymbolicDataModel(schema=schema, name=self.name)

Or

Bases: Operation

Perform a logical Or operation between data models.

Source code in synalinks/src/ops/json.py

class Or(Operation):
    """Perform a logical `Or` operation between data models."""

    def __init__(
        self,
        name=None,
        description=None,
    ):
        super().__init__(
            name=name,
            description=description,
        )

    async def call(self, x1, x2):
        if x1 and x2:
            value = concatenate_json(x1.json(), x2.json())
            schema = concatenate_schema(x1.schema(), x2.schema())
            return JsonDataModel(value=value, schema=schema, name=self.name)
        elif x1 and not x2:
            return JsonDataModel(value=x1.json(), schema=x1.schema(), name=self.name)
        elif not x1 and x2:
            return JsonDataModel(value=x2.json(), schema=x2.schema(), name=self.name)
        else:
            return None

    async def compute_output_spec(self, x1, x2):
        return SymbolicDataModel(schema=x1.schema(), name=self.name)

OutMask

Bases: Operation

Mask specific fields of a data model.

Source code in synalinks/src/ops/json.py

class OutMask(Operation):
    """Mask specific fields of a data model."""

    def __init__(
        self,
        mask=None,
        recursive=True,
        name=None,
        description=None,
    ):
        super().__init__(
            name=name,
            description=description,
        )
        self.mask = mask
        self.recursive = recursive

    async def call(self, x):
        value = out_mask_json(x.json(), mask=self.mask, recursive=self.recursive)
        schema = out_mask_schema(x.schema(), mask=self.mask, recursive=self.recursive)
        return JsonDataModel(value=value, schema=schema, name=self.name)

    async def compute_output_spec(self, x):
        schema = out_mask_schema(x.schema(), mask=self.mask, recursive=self.recursive)
        return SymbolicDataModel(schema=schema, name=self.name)

Prefix

Bases: Operation

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

Source code in synalinks/src/ops/json.py

class Prefix(Operation):
    """Add a prefix to **all** the data model fields (non-recursive)."""

    def __init__(
        self,
        prefix=None,
        name=None,
        description=None,
    ):
        super().__init__(
            name=name,
            description=description,
        )
        self.prefix = prefix

    async def call(self, x):
        value = prefix_json(x.json(), self.prefix)
        schema = prefix_schema(x.schema(), self.prefix)
        return JsonDataModel(value=value, schema=schema, name=self.name)

    async def compute_output_spec(self, x):
        schema = prefix_schema(x.schema(), self.prefix)
        return SymbolicDataModel(schema=schema, name=self.name)

Suffix

Bases: Operation

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

Source code in synalinks/src/ops/json.py

class Suffix(Operation):
    """Add a suffix to **all** the data model fields (non-recursive)."""

    def __init__(
        self,
        suffix=None,
        name=None,
        description=None,
    ):
        super().__init__(
            name=name,
            description=description,
        )
        self.suffix = suffix

    async def call(self, x):
        value = suffix_json(x.json(), self.suffix)
        schema = suffix_schema(x.schema(), self.suffix)
        return JsonDataModel(value=value, schema=schema, name=self.name)

    async def compute_output_spec(self, x):
        schema = suffix_schema(x.schema(), self.suffix)
        return SymbolicDataModel(schema=schema, name=self.name)

concat(x1, x2, name=None, description=None) async

Concatenate two data models together.

Concatenation consist in creating a new data model containing all the elements of the two inputs into a new one. Each field name is made unique if needed by adding a numerical suffix _n, with n being an incremental integer.

This operation is implemented in the + Python operator.

If any of the data models used is a metaclass or symbolic data model the output is a symbolic data model.

If any of the inputs is None, then an exception is raised.

If the keys are used more than once, a numerical suffix is added.

Table:

x1	x2	Concat (+)
x1	x2	x1 + x2
x1	None	Exception
None	x2	Exception
None	None	Exception

Parameters:

Name	Type	Description	Default
x1	JsonDataModel | SymbolicDataModel	the first input data model.	required
x2	JsonDataModel | SymbolicDataModel	the second input data model.	required
name	str	Optional. The name of the operation.	None
description	str	Optional. The description of the operation.	None

Returns:

Type	Description
JsonDataModel | SymbolicDataModel	The resulting data model

Source code in synalinks/src/ops/json.py

@synalinks_export(["synalinks.ops.concatenate", "synalinks.ops.json.concatenate"])
async def concat(x1, x2, name=None, description=None):
    """Concatenate two data models together.

    Concatenation consist in creating a new data model containing
    all the elements of the two inputs into a new one.
    Each field name is made unique if needed by adding a numerical suffix `_n`,
    with `n` being an incremental integer.

    This operation is implemented in the `+` Python operator.

    If any of the data models used is a metaclass or symbolic data model
    the output is a symbolic data model.

    If any of the inputs is None, then an exception is raised.

    If the keys are used more than once, a numerical suffix is added.

    Table:

    | `x1`   | `x2`   | Concat (`+`)      |
    | ------ | ------ | ----------------- |
    | `x1`   | `x2`   | `x1 + x2`         |
    | `x1`   | `None` | `Exception`       |
    | `None` | `x2`   | `Exception`       |
    | `None` | `None` | `Exception`       |

    Args:
        x1 (JsonDataModel | SymbolicDataModel): the first input data model.
        x2 (JsonDataModel | SymbolicDataModel): the second input data model.
        name (str): Optional. The name of the operation.
        description (str): Optional. The description of the operation.

    Returns:
        (JsonDataModel | SymbolicDataModel): The resulting data model
    """
    if any_symbolic_data_models(x1, x2):
        return await Concat(
            name=name,
            description=description,
        ).symbolic_call(x1, x2)
    return await Concat(
        name=name,
        description=description,
    )(x1, x2)

factorize(x, name=None, description=None) async

Factorize a data model by grouping similar properties together.

Factorization consist in grouping the same properties into lists. The property key of the resulting grouped property is changed to its plural form. For example action become actions, or query become queries.

If the data models used is a metaclass or symbolic data model the output is a symbolic data model.

This operation is implemented in .factorize()

Parameters:

Name	Type	Description	Default
x	JsonDataModel | SymbolicDataModel	the input data model.	required
name	str	Optional. The name of the operation.	None
description	str	Optional. The description of the operation.	None

Returns:

Type	Description
JsonDataModel | SymbolicDataModel	The resulting data model.

Source code in synalinks/src/ops/json.py

@synalinks_export(["synalinks.ops.factorize", "synalinks.ops.json.factorize"])
async def factorize(x, name=None, description=None):
    """Factorize a data model by grouping similar properties together.

    Factorization consist in grouping the same properties into lists.
    The property key of the resulting grouped property is changed to its plural form.
    For example `action` become `actions`, or `query` become `queries`.

    If the data models used is a metaclass or symbolic data model
    the output is a symbolic data model.

    This operation is implemented in `.factorize()`

    Args:
        x (JsonDataModel | SymbolicDataModel): the input data model.
        name (str): Optional. The name of the operation.
        description (str): Optional. The description of the operation.

    Returns:
        (JsonDataModel | SymbolicDataModel): The resulting data model.
    """
    if any_symbolic_data_models(x):
        return await Factorize(
            name=name,
            description=description,
        ).symbolic_call(x)
    return await Factorize(
        name=name,
        description=description,
    )(x)

in_mask(x, mask=None, recursive=True, name=None, description=None) async

Keep specific fields of a data model.

In masking consists in keeping the properties that match with the keys given in the mask. The masking process ignores the numerical suffixes that could be added by other operations.

If the data models used is a metaclass or symbolic data model the output is a symbolic data model.

Parameters:

Name	Type	Description	Default
x	JsonDataModel | SymbolicDataModel	the input data model	required
mask	list	the input mask (list of keys)	None
recursive	bool	Whether or not to keep recursively for nested objects (default True).	True
name	str	Optional. The name of the operation.	None
description	str	Optional. The description of the operation.	None

Returns:

Type	Description
JsonDataModel | SymbolicDataModel	The resulting data model.

Source code in synalinks/src/ops/json.py

@synalinks_export(["synalinks.ops.in_mask", "synalinks.ops.json.in_mask"])
async def in_mask(x, mask=None, recursive=True, name=None, description=None):
    """Keep specific fields of a data model.

    In masking consists in keeping the properties that match with the keys given
    in the mask. The masking process ignores the numerical suffixes that could be added
    by other operations.

    If the data models used is a metaclass or symbolic data model
    the output is a symbolic data model.

    Args:
        x (JsonDataModel | SymbolicDataModel): the input data model
        mask (list): the input mask (list of keys)
        recursive (bool): Whether or not to keep
            recursively for nested objects (default True).
        name (str): Optional. The name of the operation.
        description (str): Optional. The description of the operation.

    Returns:
        (JsonDataModel | SymbolicDataModel): The resulting data model.
    """
    if x is None:
        return x
    if mask is None:
        raise ValueError("You should specify the `mask` argument")
    if any_symbolic_data_models(x):
        return await InMask(
            mask=mask,
            recursive=recursive,
            name=name,
            description=description,
        ).symbolic_call(x)
    return await InMask(
        mask=mask,
        recursive=recursive,
        name=name,
        description=description,
    )(x)

logical_and(x1, x2, name=None, description=None) async

Perform a logical And operation between two data models.

If one of the inputs is None, then this operation output None. If both inputs are provided, the output is a concatenation of the two given data models.

This operation is implemented in the Python & operator.

If any of the data models used is a metaclass or symbolic data model the output is a symbolic data model.

Table:

x1	x2	Logical And (&)
x1	x2	x1 + x2
x1	None	None
None	x2	None
None	None	None

Parameters:

Name	Type	Description	Default
x1	JsonDataModel | SymbolicDataModel	The first input data model.	required
x2	JsonDataModel | SymbolicDataModel	The second input data model.	required
name	str	Optional. The name of the operation.	None
description	str	Optional. The description of the operation.	None

Returns:

Type	Description
JsonDataModel | SymbolicDataModel | None	The resulting data model or None if the condition is not met.

Source code in synalinks/src/ops/json.py

async def logical_and(x1, x2, name=None, description=None):
    """Perform a logical `And` operation between two data models.

    If one of the inputs is `None`, then this operation output `None`.
    If both inputs are provided, the output is a concatenation
    of the two given data models.

    This operation is implemented in the Python `&` operator.

    If any of the data models used is a metaclass or symbolic data model
    the output is a symbolic data model.

    Table:

    | `x1`   | `x2`   | Logical And (`&`) |
    | ------ | ------ | ----------------- |
    | `x1`   | `x2`   | `x1 + x2`         |
    | `x1`   | `None` | `None`            |
    | `None` | `x2`   | `None`            |
    | `None` | `None` | `None`            |

    Args:
        x1 (JsonDataModel | SymbolicDataModel): The first input data model.
        x2 (JsonDataModel | SymbolicDataModel): The second input data model.
        name (str): Optional. The name of the operation.
        description (str): Optional. The description of the operation.

    Returns:
        (JsonDataModel | SymbolicDataModel | None): The resulting data model or
            None if the condition is not met.
    """
    if any_symbolic_data_models(x1, x2):
        return await And(
            name=name,
            description=description,
        ).symbolic_call(x1, x2)
    return await And(
        name=name,
        description=description,
    )(x1, x2)

logical_or(x1, x2, name=None, description=None) async

Perform a logical Or between two data models.

If one of the input is None, then output the other one. If both inputs are provided, the output is a concatenation of the two given data models.

If any of the data models used is a metaclass or symbolic data model the output is a symbolic data model.

This operation is implemented in the Python | operator.

Table:

x1	x2	Logical Or (|)
x1	x2	x1 + x2
x1	None	x1
None	x2	x2
None	None	None

Parameters:

Name	Type	Description	Default
x1	JsonDataModel | SymbolicDataModel	The first input data model.	required
x2	JsonDataModel | SymbolicDataModel	The second input data model.	required
name	str	Optional. The name of the operation.	None
description	str	Optional. The description of the operation.	None

Returns:

Type	Description
JsonDataModel | SymbolicDataModel | None	The resulting data model or None if the condition is not met.

Source code in synalinks/src/ops/json.py

async def logical_or(x1, x2, name=None, description=None):
    """Perform a logical `Or` between two data models.

    If one of the input is `None`, then output the other one.
    If both inputs are provided, the output is a concatenation
    of the two given data models.

    If any of the data models used is a metaclass or symbolic data model
    the output is a symbolic data model.

    This operation is implemented in the Python `|` operator.

    Table:

    | `x1`   | `x2`   | Logical Or (`|`) |
    | ------ | ------ | ---------------- |
    | `x1`   | `x2`   | `x1 + x2`        |
    | `x1`   | `None` | `x1`             |
    | `None` | `x2`   | `x2`             |
    | `None` | `None` | `None`           |

    Args:
        x1 (JsonDataModel | SymbolicDataModel): The first input data model.
        x2 (JsonDataModel | SymbolicDataModel): The second input data model.
        name (str): Optional. The name of the operation.
        description (str): Optional. The description of the operation.

    Returns:
        (JsonDataModel | SymbolicDataModel | None): The resulting data model or
            None if the condition is not met.
    """
    if any_symbolic_data_models(x1, x2):
        return await Or(
            name=name,
            description=description,
        ).symbolic_call(x1, x2)
    return await Or(
        name=name,
        description=description,
    )(x1, x2)

out_mask(x, mask=None, recursive=True, name=None, description=None) async

Mask specific fields of a data model.

Out masking consist in removing the properties that match with the keys given in the mask. The masking process ignore the numerical suffixes that could be added by other operations.

If the data models used is a metaclass or symbolic data model the output is a symbolic data model.

Parameters:

Name	Type	Description	Default
x	JsonDataModel | SymbolicDataModel	the input data model.	required
mask	list	the input mask (list of keys).	None
recursive	bool	Whether or not to remove recursively for nested objects (default True).	True
name	str	Optional. The name of the operation.	None
description	str	Optional. The description of the operation.	None

Returns:

Type	Description
JsonDataModel | SymbolicDataModel	The resulting data model.

Source code in synalinks/src/ops/json.py

@synalinks_export(["synalinks.ops.out_mask", "synalinks.ops.json.out_mask"])
async def out_mask(x, mask=None, recursive=True, name=None, description=None):
    """Mask specific fields of a data model.

    Out masking consist in removing the properties that match with the keys given
    in the mask. The masking process ignore the numerical suffixes that could be added
    by other operations.

    If the data models used is a metaclass or symbolic data model
    the output is a symbolic data model.

    Args:
        x (JsonDataModel | SymbolicDataModel): the input data model.
        mask (list): the input mask (list of keys).
        recursive (bool): Whether or not to remove
            recursively for nested objects (default True).
        name (str): Optional. The name of the operation.
        description (str): Optional. The description of the operation.

    Returns:
        (JsonDataModel | SymbolicDataModel): The resulting data model.
    """
    if x is None:
        return x
    if mask is None:
        raise ValueError("You should specify the `mask` argument")
    if any_symbolic_data_models(x):
        return await OutMask(
            mask=mask,
            recursive=recursive,
            name=name,
            description=description,
        ).symbolic_call(x)
    return await OutMask(
        mask=mask,
        recursive=recursive,
        name=name,
        description=description,
    )(x)

prefix(x, prefix=None, name=None, description=None) async

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

If the data models used is a metaclass or symbolic data model the output is a symbolic data model.

Parameters:

Name	Type	Description	Default
x	JsonDataModel | SymbolicDataModel	the input data model	required
prefix	str	the prefix to add.	None
name	str	Optional. The name of the operation.	None
description	str	Optional. The description of the operation.	None

Returns:

Type	Description
JsonDataModel | SymbolicDataModel	The resulting data model.

Source code in synalinks/src/ops/json.py

@synalinks_export(["synalinks.ops.prefix", "synalinks.ops.json.prefix"])
async def prefix(x, prefix=None, name=None, description=None):
    """Add a prefix to **all** the data model fields (non-recursive).

    If the data models used is a metaclass or symbolic data model
    the output is a symbolic data model.

    Args:
        x (JsonDataModel | SymbolicDataModel): the input data model
        prefix (str): the prefix to add.
        name (str): Optional. The name of the operation.
        description (str): Optional. The description of the operation.

    Returns:
        (JsonDataModel | SymbolicDataModel): The resulting data model.
    """
    if x is None:
        return x
    if prefix is None:
        raise ValueError("You should specify the `prefix` argument")
    if any_symbolic_data_models(x):
        return await Prefix(
            prefix=prefix,
            name=name,
            description=description,
        ).symbolic_call(x)
    return await Prefix(
        prefix=prefix,
        name=name,
        description=description,
    )(x)

suffix(x, suffix=None, name=None, description=None) async

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

If the data models used is a metaclass or symbolic data model the output is a symbolic data model.

Parameters:

Name	Type	Description	Default
x	JsonDataModel | SymbolicDataModel	the input data model	required
suffix	str	the suffix to add.	None
name	str	Optional. The name of the operation.	None
description	str	Optional. The description of the operation.	None

Returns:

Type	Description
JsonDataModel | SymbolicDataModel	The resulting data model

Source code in synalinks/src/ops/json.py

@synalinks_export(["synalinks.ops.suffix", "synalinks.ops.json.suffix"])
async def suffix(x, suffix=None, name=None, description=None):
    """Add a suffix to **all** the data model fields (non-recursive).

    If the data models used is a metaclass or symbolic data model
    the output is a symbolic data model.

    Args:
        x (JsonDataModel | SymbolicDataModel): the input data model
        suffix (str): the suffix to add.
        name (str): Optional. The name of the operation.
        description (str): Optional. The description of the operation.

    Returns:
        (JsonDataModel | SymbolicDataModel): The resulting data model
    """
    if x is None:
        return x
    if suffix is None:
        raise ValueError("You should specify the `suffix` argument")
    if any_symbolic_data_models(x):
        return await Suffix(
            suffix=suffix,
            name=name,
            description=description,
        ).symbolic_call(x)
    return await Suffix(
        suffix=suffix,
        name=name,
        description=description,
    )(x)