Schema Validation

Cloudomation allows you to define Schemas against which you can validate data.

Use Cases

Use Schemas to

• validate input data for flows or connections,
• validate output data to ensure you received all necessary data for further processing,
• coerce your data and ensure it is transformed to the correct data types for further processing.

Concept

In Cloudomation you can define a Schema both via the user interface with Create or directly inside a Flow. Afterwards, you can validate or coerce your data against a given Schema.

A Schema allows you to define e.g.:

• keys to be validated,
• data type,
• data format,
• default values if a key is not present,
• required keys.

Upon successful validation or coersion the (coerced) data is returned.

note

Currently, JSON schema validation with draft 7 formatting is implemented.

warning

When validation of the data fails a flow_api.exceptions.SchemaValidationError exception is raised. You can handle this exception as described in Exceptions

warning

When the schema itself is invalid, a flow_api.exceptions.SchemaError exception is raised. You can handle this exception as described in Exceptions

Defining a schema

Schemas can be defined using the user interface of Cloudomation or within a flow script.

Create a Schema in the user interface

You can simply create a new Schema in the user interface:

The buttons to create a schema

Write your schema in the editor.

Create a Schema inside flows

You can save a schema for later validation or coersion within a flow with the following syntax:
system.schema(<schema-name>).save(value=<dict with schema definition>)

See the following example:

import flow_api

def handler(system: flow_api.System, this: flow_api.Execution):
    system.schema('my schema').save(value={
        'type': 'object',
        'properties': {
            'some_id': {
                'type': 'string',
                'format': 'uuid',},
            'name': {
                'type': 'string',},
            'date': {
                'type': 'string',
                'format': 'datetime',},
            'number': {
                'type': 'number',
                'default': 2,},
        },
        'required': [
            'some_id',
            'name',
        ],
    })
    return this.success('all done')

Create a Schema with a wrapper

You can create and also update a schema dynamically using the validate wrapper. Setting the wrapper's parameter mode to learn, allows you to generate a json schema from the input_value or output_value of an Execution. For more information on this, refer to validate.

Validating data

You can validate your data with the schema's method validate(data=<my_data>).

import flow_api

def handler(system: flow_api.System, this: flow_api.Execution):
    my_data = {
        'some_id': '12345678-1abc-2cde-3def-987654321cba',
        'name': 'giraffe',
        'date': '2020-12-24 20:20',
        'additional': 'additional value',
    }

    validated_data = system.schema('my schema').validate(data=my_data)

    return this.success('all done')

Validate a Schema with a wrapper

You can validate the input_value or output_value of an Execution using the validate wrapper. For more information on this, refer to validate.

Coercing data

If you want to coerce data, either convert compatible types or add default values for missing keys,
then use validate(data=<my_data>, coerce=True) or its alias coerce(data=<my_data>).

import flow_api

def handler(system: flow_api.System, this: flow_api.Execution):
    my_data = {
        'name': '9999',
        'some_id': '12345678-1abc-2cde-3def-987654321cba',
    }

    coerced_data = system.schema('my schema').coerce(data=my_data)
    coerced_data = system.schema('my schema').validate(data=my_data, coerce=True)

    return this.success('all done')

Handling Validation Errors

If a schema validation or coersion fails the exception flow_api.exceptions.SchemaValidationError is thrown. You have to catch the error and can decide what to do with the situation. Either, you add further processing, stop execution of the flow or simply output a warning and ignore the failed validation.

In this example, the flow's input-data is validated and the error is written to the log:

import flow_api

def handler(system: flow_api.System, this: flow_api.Execution):
    inputs = this.get('input_value')

    try:
        validated_data = system.schema('my schema').validate(data=inputs)
    except flow_api.exceptions.SchemaValidationError as ex:
        this.log(validation_error=repr(ex))

    # here follows further code

    return this.success('all done')