Create a simple Job (no hooks / rules)

from impulsare_job import Writer


writer = Writer('/etc/impulsare/config.yml')
writer.set_prop('name', 'My Job')
writer.set_prop('input', 'csv')
writer.set_prop('input_parameters', {'delimiter': 'csv'})
writer.set_prop('output', 'sql')
writer.set_prop('output_parameters', {'db': 'test'})
job = writer.save()

Update a Job

from impulsare_job import Writer


# Lets assume the job id = 1
writer = Writer('/etc/impulsare/config.yml', 'My Job')
job = writer.get_job()
print(job.name)
# Output: 'My Job'

# Set the job to Inactive
writer.set_prop('active', False)
writer.save()

Verify if a hook exists, else add it

# .... continuation of code above
if not writer.hooks_writer.hook_exists('test'):
    writer.hooks_writer.add_hook(name='upload_file', method='upload_file', when='after_process')

Allowed properties for hooks:

{
    'name': str, # required
    'method': str, # required
    'when': str, # required
    'description': str,
    'active': bool, # Default : True
    'priority': int # Default: 1
}

Other methods:

• hooks_writer.get_hooks

• hooks_writer.del_hook

There is no method update, to update a hook, delete it then recreate it.

Verify if a field exists, else update it and add a transformation rule

Warning : a field is identified by its output value that must be unique (we can’t send two values for the same field while we can use the same input field for various output).

Field

# .... continuation of code above
if writer.fields_writer.field_exists('firstname'):
    writer.fields_writer.del_field('firstname')

writer.fields_writer.add_field(input='first_name', output='firstname')

Allowed properties for fields:

{
    'input': str, # required
    'output': str, # required
}

Other methods:

• fields_writer.get_field

• fields_writer.get_fields

There is no method update, to update a field, delete it then recreate it.

Add a rule

writer.fields_writer.add_rule(output_field='firstname', name='uppercase', method='uppercase')

Allowed properties for rules:

{
    'name': str, # required
    'method': str, # required
    'description': str,
    'active': bool, # Default : True
    'params': list,
    'blocking': bool, # Default : False
    'priority': int # Default: 1
}

Other methods:

• fields_writer.del_rule

• fields_writer.get_rules

• fields_writer.rule_exists

There is no method update, to update a rule, delete it then recreate it.

Retrieve a Job, its hooks and fields

from impulsare_job import Reader


Reader = Reader('/etc/impulsare/config.yml', 'My Job')
job = Reader.get_job()
hooks = Reader.get_hooks()
fields = Reader.get_fields() # Get rules for first field : rules = fields[0].rules

Don’t check if table exists on each model but do it on app installation

To get the DB :

from impulsare_job import models


db = models.get_db('/etc/impulsare/config.yml')
db.create_tables([models.Job, models.Hook, models.Rule])

Refactor writer

To have a class for hooks, and another for rules.