rubenf•8mo ago

Workflows as Code | Windmill

We are releasing in beta Workflow as code for Python and Typescript. No more excuse to use Airflow or Prefect: https://www.windmill.dev/docs/core_concepts/workflows_as_code

from wmill import task

import pandas as pd
import numpy as np

# The pin is important since task is a decorator available only from 1.286.2

#extra_requirements:
#wmill>=1.286.2


##You can specify tag to run the task on a specific type of worker
@task(tag="highmem")
def heavy_compute(n: int):
    df = pd.DataFrame(np.random.randn(100, 4), columns=list('ABCD'))
    return df.sum().sum()


@task
def send_result(res: int, email: str):
    # logs of the subtask are available in the main task logs
    print(f"Sending result {res} to {email}")
    return "OK"

def main(n: int):
    l = []

    # to run things in parallel, simply use multiprocessing Pool map instead: https://docs.python.org/3/library/multiprocessing.html
    for i in range(n):
        l.append(heavy_compute(i))
    print(l)
    return send_result(sum(l), "example@example.com")

from wmill import task

import pandas as pd
import numpy as np

# The pin is important since task is a decorator available only from 1.286.2

#extra_requirements:
#wmill>=1.286.2


##You can specify tag to run the task on a specific type of worker
@task(tag="highmem")
def heavy_compute(n: int):
    df = pd.DataFrame(np.random.randn(100, 4), columns=list('ABCD'))
    return df.sum().sum()


@task
def send_result(res: int, email: str):
    # logs of the subtask are available in the main task logs
    print(f"Sending result {res} to {email}")
    return "OK"

def main(n: int):
    l = []

    # to run things in parallel, simply use multiprocessing Pool map instead: https://docs.python.org/3/library/multiprocessing.html
    for i in range(n):
        l.append(heavy_compute(i))
    print(l)
    return send_result(sum(l), "example@example.com")

Flows are not the only way to write distributed programs that execute distinct jobs. Another approach is to write a program that defines the jobs and their dependencies, and then execute that program. This is known as workflows as code.

7 Replies

fr3fou•7mo ago

hey, are there plans on making a typescript decorator? the API for ts does seem annoying to use

rubenf•7mo ago

issue is typescript doesn't allow decorator at top level, it require to declare a class which is also annoying

fr3fou•7mo ago

damn, i forgot would you be interested in supporting alternative class based approach though? @flow and @task decorators seem like a neat idea

invakid404•7mo ago

fwiw you could also use arrow functions and wrap it once in task() which would be much less annoying afaict the current implementation of task() doesn't require you to make a new function every time

fr3fou•7mo ago

import { task } from 'windmill-client';

export const hello = task((name: string) => {
    return 'Hello:' + name;
})

export async function main() {
    //It's the function itself that needs to be wrapped with task, and it's always a promise even
    await task('BAR');
    return task('FOO');
}

import { task } from 'windmill-client';

export const hello = task((name: string) => {
    return 'Hello:' + name;
})

export async function main() {
    //It's the function itself that needs to be wrapped with task, and it's always a promise even
    await task('BAR');
    return task('FOO');
}

yeah that doesn't look that bad small update - you can't do this

InternalErr: Internal: Could not add completed job 018eb467-59be-90bb-ee5d-5ddfa1a0f5d4: error returned from database: invalid input syntax for type json

seems like the individual tasks have to use the function hello(...) syntax and therefore the windmill.task() call must be done in main() me and @invakid404 also did some further tests and for some reason the tasks can't destructure objects inline:

import { task } from 'windmill-client';

type HelloArgs = {
  id: string
  name: string
}

async function hello({ id, name }: HelloArgs) { 
    return 'Hello:' + name + id;
}

export async function main() {
    await task(hello)({id: 'asdf', name: 'fefe'});
}

import { task } from 'windmill-client';

type HelloArgs = {
  id: string
  name: string
}

async function hello({ id, name }: HelloArgs) { 
    return 'Hello:' + name + id;
}

export async function main() {
    await task(hello)({id: 'asdf', name: 'fefe'});
}

Error: Job 018eb466-87ee-1607-3b3c-cadb4504bab6 was not successful: {"name":"ExecutionErr","message":"error during execution of the script:\nmain function was not findable (expected to find 'export function main(...)'"}

actually, the example i gave isn't related to the destructuring 🤔 we did however fix an error regarding "right hand destructuring assignment" by removing the inline arg destructuring

rubenf•7mo ago

@fr3fou I don't know if you saw but the task code is open-source: https://github.com/windmill-labs/windmill/blob/main/typescript-client/client.ts#L188

GitHub

windmill/typescript-client/client.ts at main · windmill-labs/windmill

Open-source developer platform to turn scripts into workflows and UIs. Fastest workflow engine (5x vs Airflow). Open-source alternative to Airplane and Retool. - windmill-labs/windmill

rubenf•7mo ago

it's possible to contribute with your own syntax sugar