Scheduled Functions

​

Overview

Some jobs run on a clock, not on a request — refreshing a table of exchange rates, recomputing a daily rollup, warming a cache, syncing an external source. These are ordinary Python jobs, but they need a stable environment with the right libraries and credentials, and something reliable to trigger them.

A scheduled function is a Compute Function that Chalk invokes for you on a recurring schedule. You write the function once, declare a schedule in the decorator, and deploy it — Chalk runs it on the platform at the cadence you specify, with the custom image, resources, secrets, and retry behavior you configured. There is no separate cron server to operate.

​

Define the image

Bake your dependencies into an image so each run starts ready to go. Anything your job imports — data clients, model libraries — belongs here:

import chalkcompute

image = chalkcompute.Image.debian_slim().run_commands(
    "pip install requests",
)

​

Write the scheduled function

Add the schedule argument to the @chalkcompute.function decorator. A scheduled function takes no arguments — it is invoked by the platform, so there is no caller to pass inputs. Everything it needs, it fetches or computes itself:

import chalkcompute

@chalkcompute.function(
    name="refresh-exchange-rates",
    schedule="0 * * * *",                      # every hour, on the hour
    image=image,
    secrets=[chalkcompute.Secret.from_env("EXCHANGE_API_KEY")],
    cpu="1",
    memory="1Gi",
    retries=3,
)
def refresh_exchange_rates() -> None:
    import os
    import requests

    resp = requests.get(
        "https://api.example.com/v1/rates",
        params={"base": "USD"},
        headers={"Authorization": f"Bearer {os.environ['EXCHANGE_API_KEY']}"},
    )
    rates = resp.json()["rates"]
    # write the latest rates to your store
    ...

The custom image, cpu / memory requests, and secrets configure the Scaling Group that hosts the function; schedule and retries govern how and when the platform invokes it.

Scheduled jobs often do heavier work — nightly batch inference, periodic re-scoring of a dataset. Request a GPU the same way you would for any function:

@chalkcompute.function(
    name="nightly-batch-inference",
    schedule="0 3 * * *",      # daily at 03:00 UTC
    image=image,
    gpu="nvidia-l4",
    cpu="4",
    memory="16Gi",
)
def nightly_batch_inference() -> None:
    ...

​

Deploy it

Deploying registers the function and its schedule with the platform. Run the script:

uv run python refresh_exchange_rates.py

From this point on, Chalk invokes refresh_exchange_rates on its schedule. No further action is required to keep it running.

​

Schedule syntax

The schedule argument accepts either a crontab expression or a Chalk duration string. Use whichever reads more clearly for your cadence. All schedules are evaluated in UTC.

Crontab — five space-separated fields (minute hour day-of-month month day-of-week). Standard syntax is supported: *, lists (1,15), ranges (1-5), and steps (*/10).

Duration shorthand — a single interval, for simple cadences. Supported units are minutes (≤ 60), hours (≤ 24), and days (= 1), written as one value:

Shorthand intervals are aligned to clock boundaries, not to your deployment time: "6h" runs at 00:00, 06:00, 12:00, and 18:00 UTC. Anything longer or compound ("90m", "1h30m", "2d") must be written as a crontab expression.

@chalkcompute.function(schedule="0 2 * * *")   # nightly at 02:00 UTC
def nightly_rollup() -> None:
    ...

@chalkcompute.function(schedule="15m")         # every 15 minutes
def warm_cache() -> None:
    ...

​

Rules for scheduled functions

Because a scheduled function is invoked by the platform rather than a caller, two constraints apply and are enforced when the function is defined:

• No inputs. A scheduled function must take no arguments. Have it read its own inputs from your sources.
• Not a generator. A scheduled run has no caller to consume yielded values, so a scheduled function cannot yield. Return None (or a value) instead.

Everything else about a Compute Function still applies — custom images, resource requests, secrets, and retries all work exactly as they do for on-demand functions.

​

Observe runs

Each invocation appears in the Chalk dashboard with its status, logs, and resource usage, just like a manually invoked function. Emit logs from the body to make each run observable:

from chalk import chalk_logger

@chalkcompute.function(name="refresh-exchange-rates", schedule="1h")
def refresh_exchange_rates() -> None:
    chalk_logger.info("refresh started")
    rates = ...  # fetch the latest rates and write them to your store
    chalk_logger.info("refreshed %s rates", len(rates))