Skip to main content

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.

info

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')

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')

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')