zenaton / zenaton-python

🐍 Python library to run and orchestrate background jobs with Zenaton Workflow Engine

⚠️ This repository is abandoned.

Easy Asynchronous Jobs Manager for Developers
Explore the docs »
Website · Examples in Python · Tutorial in Python

Zenaton library for Python

Zenaton helps developers to easily run, monitor and orchestrate background jobs on your workers without managing a queuing system. In addition to this, a monitoring dashboard shows you in real-time tasks executions and helps you to handle errors.

The Zenaton library for Python lets you code and launch tasks using Zenaton platform, as well as write workflows as code. You can sign up for an account on Zenaton and go through the tutorial in python.

Requirements

This package has been tested with Python 3.5.

Python Documentation

You can find all details on Zenaton's website.

Table of contents

• Zenaton library for Python
  • Requirements
  • Python Documentation
  • Getting started
    • Installation
      • Install the Zenaton Agent
      • Install the library
      • Framework integration
    • Quick start
      • Client Initialization
      • Executing a background job
    • Orchestrating background jobs
      • Using workflows
  • Getting help
    • Theorical Examples
    • Real-life Examples
  • Contributing
  • License
  • Code of Conduct

Getting started

Installation

Install the Zenaton Agent

To install the Zenaton agent, run the following command:

curl https://install.zenaton.com/ | sh

Then, you need your agent to listen to your application. To do this, you need your Application ID and API Token. You can find both on your Zenaton account.

zenaton listen --app_id=YourApplicationId --api_token=YourApiToken --app_env=YourApplicationEnv --boot=boot.py

Install the library

To add the latest version of the library to your project, run the following command:

pip install zenaton

Framework integration

If you are using Django, please refer to our dedicated documentation to get started:

• Getting started with Django

Quick start

Client Initialization

To start, you need to initialize the client. To do this, you need your Application ID and API Token. You can find both on your Zenaton account.

Then, initialize your Zenaton client:

from zenaton.client import Client

Client(your_app_id, your_api_token, your_app_env)

Executing a background job

A background job in Zenaton is a class implementing the Zenaton.abstracts.task.Task interface.

Let's start by implementing a first task printing something, and returning a value:

import random

from zenaton.abstracts.task import Task
from zenaton.traits.zenatonable import Zenatonable

class HelloWorldTask(Task, Zenatonable):

    def handle(self):
        print('Hello World\n')
        return random.randint (0, 1)

Now, when you want to run this task as a background job, you need to do the following:

HelloWorldTask().dispatch()

That's all you need to get started. With this, you can run many background jobs. However, the real power of Zenaton is to be able to orchestrate these jobs. The next section will introduce you to job orchestration.

Orchestrating background jobs

Job orchestration is what allows you to write complex business workflows in a simple way. You can execute jobs sequentially, in parallel, conditionally based on the result of a previous job, and you can even use loops to repeat some tasks.

We wrote about some use-cases of job orchestration, you can take a look at these articles to see how people use job orchestration.

Using workflows

A workflow in Zenaton is a class implementing the Zenaton.abstracts.workflow.Workflow interface.

We will implement a very simple workflow:

First, it will execute the HelloWorld task. The result of the first task will be used to make a condition using an if statement. When the returned value will be greater than 0, we will execute a second task named FinalTask. Otherwise, we won't do anything else.

One important thing to remember is that your workflow implementation must be idempotent. You can read more about that in our documentation.

The implementation looks like this:

from tasks.hello_world_task import HelloWorldTask
from tasks.final_task import FinalTask

from zenaton.abstracts.workflow import Workflow
from zenaton.traits.zenatonable import Zenatonable

class MyFirstWorkflow(Workflow, Zenatonable):

    def handle(self):

        n = HelloWorldTask().execute()

        if n > 0:
            FinalTask().execute()

Now that your workflow is implemented, you can execute it by calling the dispatch method:

MyFirstWorkflow().dispatch()

If you really want to run this example, you will need to implement the FinalTask task.

There are many more features usable in workflows in order to get the orchestration done right. You can learn more in our documentation.

Getting help

Need help? Feel free to contact us by chat on Zenaton.

Found a bug? You can open a GitHub issue.

Theorical Examples

Python examples repo

Real-life Examples

Triggering An Email After 3 Days of Cold Weather (Medium Article, Source Code)

Contributing

Bug reports and pull requests are welcome on GitHub here. This project is intended to be a safe, welcoming space for collaboration, and contributors are expected to adhere to the Contributor Covenant code of conduct.

Testing

To test your changes before sending a pull request, first install the tests requirements:

pip install '.[test]'

Then run PyTest:

pytest

License

The package is available as open source under the terms of the MIT License.

Code of Conduct

Everyone interacting in the zenaton-Python project’s codebases, issue trackers, chat rooms and mailing lists is expected to follow the code of conduct.

zenaton-python's Issues

Correct the behaviour of Event Subclasses in the Python Library

As known for some times, the when subclassing and Event class, the child Event is not handled properly by Zenaton changes has to be done, in coordination between the engine, agent and client libraries.

For more details please check:

zenaton/rfcs#19

Improve error message when a workflow cannot be initialized from its serialized data

I made a mistake in my code, with a typo on are_all_subworkflows_complete (once with a s and once without ):

class MainWorkflow(Workflow, Zenatonable):
 
  def __init__(self, user_id: str, are_all_subworkflows_complete: bool = False) -> None:
    self.user_id = user_id
    self.are_all_subworkflows_complete = are_all_subworkflows_complete

  def handle(self) -> None:
    ...

  def on_event(self, event: Event) -> None:
    if isinstance(event, SubWorklfowEvent):
      self.are_all_subworkflow_complete = True  # Typo here!

When my MainWorkflow was run, then I got the error:

File "/root/.zenaton/lib/worker-0.7.3/priv/python/lib/zenaton/worker/v1/workflow.py", line 106, in build_workflow
return Importer(self.boot).import_workflow(self.name, Serializer(boot=self.boot, name=self.name).decode(properties))
File "/root/.zenaton/lib/worker-0.7.3/priv/python/lib/zenaton/services/importer.py", line 24, in import_workflow
return getattr(boot, name)(**args)
TypeError: __init__() got an unexpected keyword argument 'are_all_subworkflow_complete'

I lost 20 min debugging the thing. What would have helped, woud have been to get an error message like:

Workflow MainWorkflow could not be initialized because its property `are_all_subworkflow_complete` is not a valid argument of its __init__ method. Only the following properties can be passed: `are_all_subworkflows_complete` 

Error when running examples

(zenaton) [q@q-pc examples-python]$ pip freeze
certifi==2019.6.16
chardet==3.0.4
idna==2.8
python-dotenv==0.9.1
pytz==2019.1
requests==2.22.0
urllib3==1.25.3
zenaton==0.3.4
(zenaton) [q@q-pc examples-python]$ zenaton listen --boot=boot.py
Traceback (most recent call last):
 File "/home/q/.zenaton/lib/worker-0.7.8/priv/python/scripts/test_listen.py", line 14, in <module>
   import python.lib.zenaton.loader.loader as loader
 File "/home/q/.zenaton/lib/worker-0.7.8/priv/python/lib/zenaton/loader/loader.py", line 6, in <module>
   import zenaton
ModuleNotFoundError: No module named 'zenaton'

(zenaton) [q@q-pc examples-python]$

From a user using python 3.7

Raise exception if a Task is trying to send an Event that is not declared in zenaton_boot.py

Issue: in a task, I am sending an event MyEvent, but I forgot to import MyEvent in zenaton_boot.py. It seems that then Zenaton created a event from a new class 'workflow.MyEvent' on the fly instead of using 'my_module.my_workflow.MyEvent', making logic like isinstanceof(event, MyEvent) not working in the rest of my code downstream.

Solution:

• quick: Raise an exception when trying to create an event whose class is not imported in zenaton_boot.py explaining that I need to add the import
• better: have a different system for imports in zenaton_boot.py, like telling Zenaton to import all Zenaton objects in a tree (ìmport my_workflows`=> would also import everything under)

By default, the zenaton.log must no appear

Test the SDK

Is your feature request related to a problem? Please describe.
The SDK library itself has some tests (in the tests folder), however it's not easy to understand how to run them, and apparently they are not run before releasing (for instance the typo fixed in #26 got released).

Describe the solution you'd like

• Some documentation in the README.md on how to run the tests locally
• Fix all the existing tests so that they pass in multiple environments (for instance by setting the timezone in the fixtures)
• Run the tests automatically in a CI for each commit (TravisCI, CircleCI and others are free for Open Source projects).

Describe alternatives you've considered
At this stage I think we should either make those tests runnable or remove them as they give a false feeling of safety.

Use of `Wait` class outside a workflow should thrown an exception

To be clear, the test should be that the class is used within a workflow - NOT that it is processed by an agent. So locally executing a workflow is still perfectly fine.

A specific Zenaton error should be thrown and documented.

Single Task Dispatch Feature

For details have a look here => zenaton/zenaton-php#15

Task execution fails when passing an Event as task argument due likely to serialization issue of Event object

Exception comes from def import_class(self, workflow_name, class_name): where workflow_name = None, class_name= CompleteRiderTypeformEvent and self.boot = None

Add the ability to run recurrent task & workflow

Execute onStart, onSuccess, onFailure, onTimeout only when methods exist

Might not be needed on the client part.

Related to: https://github.com/zenaton/agent/issues/132

Problems in wait methods when the given day is today.

Expected Behavior:

per the discussion with @geomagilles:

if we are on Monday, 11am on the 23rd of the month. Then:

.Monday(1) should wait until next Monday at 11am
.Monday(1).At("9") should wait until next Monday at 9am
.Monday(1).At("13") should wait 2 hours only
.DayOfMonth(23) should wait for one month
.DayOfMonth(23).At("13") should wait for only 2 hours

Problem:

If it's the 23rd and you call Wait().day_of_month(23) you won't wait until the next month, but instead the wait will terminate immediately.

Specifically, I believe the problematic line is

zenaton-python/zenaton/traits/with_timestamp.py

Line 125 in c5a4596

if now > now_dup:

From just reading the code, it seems that the python library doesn't have the same problem with the weekday methods as the other libraries had/have. But I think we may have a different problem where monday().at("8") (if it's Monday at 7) will wait for a week, when it should only wait for an hour.

I think this is the relevant line for this problem (if indeed there is a problem here):

zenaton-python/zenaton/traits/with_timestamp.py

Line 94 in c5a4596

if days_ahead <= 0: # Target day already happened this week

Fix:

See zenaton/zenaton-go#7 for the fix in the go library.

Make to sure to encode query parameters in HTTP call

eg with the workflow id:

[email protected] the kill does not work.

Generate ID when scheduling things

When tasks and workflows are scheduled, we want to add a generated ID from the library.
This ID will be a canonical ID representing the user intent.

Django Support

Implement Django support, similar to Rails, Symfony and Laravel supports.

Some errors are not reported in UI

Describe the bug
in zenaton/examples-python I create a new task

import time

from workflows.asynchronous_workflow import AsynchronousWorkflow
from zenaton.abstracts.task import Task
from zenaton.traits.zenatonable import Zenatonable

class TaskWF1(Task, Zenatonable):

    def handle(self):
       AsynchronousWorkflow().dispatch()

When I dispatch this task, the error goes in zenaton.out but not in the UI

To Reproduce
See description

Expected behavior
The error should appear in Zenaton UI

Use of `execute` method outside a workflow should thrown an exception

Expected behavior of execute method does make sense only within a workflow.

To be clear, the test should be that the method is used within a workflow - NOT that it is processed by an agent. So locally executing a workflow is still perfectly fine.

A specific Zenaton error should be thrown and documented.