Skip to content

Basic Datashare worker

Install dependencies

Start by installing uv and dependencies:

curl -LsSf https://astral.sh/uv/install.sh | shuv sync --frozen --group dev

Implement the hello_user task function

As seen in the task tutorial, one of the dummiest tasks we can implement take the user: dict | None argument automatically added by Datashare to all tasks and greet that user.

The function performing this task is the following 

def hello_user(user: dict | None) -> str:
    greeting = "Hello "
    if user is None:
        user = "unknown"
    else:
        user = user["id"]
    return greeting + user

Register the hello_user task

In order to turn our function into a Datashare task, we have to register it into the app async app variable of the app.py file, using the @task decorator.

Since we won't use existing tasks, we can also perform some cleaning and get rid of them. The app.py file should hence look like this:

app.py
from icij_worker import AsyncApp
from icij_worker.app import TaskGroup

app = AsyncApp("some-app")

PYTHON_TASK_GROUP = TaskGroup(name="PYTHON")


@app.task(name="hello_user", group=PYTHON_TASK_GROUP)
def hello_user(user: dict | None) -> str:
    greeting = "Hello "
    if user is None:
        user = "unknown"
    else:
        user = user["id"]
    return greeting + user

The only thing we had to do is to use the @app.task decorator and make sure to provide it with name to bind the function to a task name and group the task in the PYTHON_TASK_GROUP = TaskGroup(name="PYTHON").

As detailed in here, using this task group ensures that when custom tasks are published for execution, they are correctly routed to your custom Python worker and not to the Java built-in workers running behind Datashare's backend.

Get rid of unused codebase

Next

Now that you have created a basic app, you can either: