Example — Python Worker

A background worker that processes messages from a Zenifra managed Valkey Queue.

Project structure

worker/
├── worker.py
└── requirements.txt

Code

`requirements.txt`

valkey

`worker.py`

import logging
import os
import time
from urllib.parse import urlparse

from valkey import Valkey

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('worker')

# Connect to the Valkey Queue over TLS using the connection copied from the console
url = urlparse(os.environ['VALKEY_URL'])
client = Valkey(
    host=url.hostname,
    port=url.port,
    username=url.username,
    password=url.password,
    ssl=True,
    ssl_check_hostname=True,
    decode_responses=True,
)

def process_message(message):
    """Process a single message."""
    logger.info(f"Processing: {message}")
    time.sleep(2)  # Simulate work
    logger.info(f"Done: {message}")
    return True

def main():
    logger.info("Worker started")
    while True:
        # Block until a message is available
        item = client.blpop('tasks', timeout=5)
        if item:
            process_message(item[1])

if __name__ == '__main__':
    main()

Console configuration

First create a Queue project and store the valkeys:// connection shown at creation. Then create the worker HTTP project with:

Field	Value
Project source	Git repository, with Git provider GitHub or Forgejo
Runtime	Python, version 3.13
Start Command	`python worker.py`
Exposure	Private - no public route
Port	`8000` (the worker does not receive HTTP requests)
Environment variable	`VALKEY_URL` with the Queue connection

Deploy

Push the code to a GitHub or Forgejo repository.
In the console, choose Git repository as Project source and GitHub or Forgejo as Git provider, and select the repository.
Select the main branch, set Deployment mode to Automatic by branch, and create the project.
Push to the branch to publish a new worker version automatically.

Push a task

# From another service or your local machine, with VALKEY_URL set
valkey-cli -u "${VALKEY_URL/valkeys:/rediss:}" --tls LPUSH tasks '{"type": "email", "to": "[email protected]"}'

A list with BLPOP is simple, but a message being processed is lost if the worker stops midway. For explicit acknowledgement and reprocessing, use Streams with consumer groups, as shown in Queues.

Expected logs

INFO:worker:Worker started
INFO:worker:Processing: {"type": "email", "to": "[email protected]"}
INFO:worker:Done: {"type": "email", "to": "[email protected]"}

Next examples

Full-stack app

On this page