Jobs

At some point, you may want to run some long-running tasks, for example, to send an email or process a file upload, that would take too long to run during a normal request-response cycle. In those cases, you can use asynchronous jobs to run those tasks in the background.

Expanse provides a simple and intuitive way to create and manage background jobs that is powered by its own message bus. This means that you must configure at least one transport for the message bus to be able to use jobs. You can find more information on how to configure transports in the message bus documentation.

Creating jobs

A job is a unit of work that can be executed asynchronously in the background. it must be defined as a class that inherits from the Job class provided by Expanse and specifies a payload type.

You can generate a new job using the make job command of the CLI:

expanse make job ProcessFileUpload

expanse make job ProcessFileUpload --sync

This will generate a new job class in the app.jobs.process_file_upload module with the following content:

from dataclasses import dataclass

from expanse.jobs.asynchronous.job import Job


@dataclass
class ProcessFileUploadPayload:

    file_path: str


class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    async def execute(self) -> None:
        file_path = payload.file_path

        # Process the file
        ...

from dataclasses import dataclass

from expanse.jobs.synchronous.job import Job


@dataclass
class ProcessFileUploadPayload:

    file_path: str


class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    def execute(self) -> None:
        file_path = self.payload.file_path

        # Process the file
        ...

Similarly to messages, Expanse supports dataclasses, Pydantic models and msgspec structs as job payloads.

Dependency injection

The execute method supports dependency injection just like route handlers or commands. For instance, if you need to access the database to process the file upload, you can inject a Session instance directly into the method:

class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    async def execute(self, storage: StorageManager) -> None:
        file = await storage.get(self.payload.file_path)

        # Process the file
        ...

Dispatching jobs

Once you have created your job, you can dispatch it using the dispatch() (or dispatch_sync()) method.

from expanse.http.upload_file import UploadFile
from expanse.jobs.job_dispatcher import JobDispatcher

from app.jobs.process_file_upload import ProcessFileUpload, ProcessFileUploadPayload


async def upload_file(
    file_: UploadFile
) -> Response:
    file_path = await file_.save("files")

    await ProcessFileUpload(
        ProcessFileUploadPayload(file_path=file_path)
    ).dispatch()

    ...

from expanse.http.upload_file import UploadFile
from expanse.jobs.job_dispatcher import JobDispatcher

from app.jobs.process_file_upload import ProcessFileUpload, ProcessFileUploadPayload


def upload_file(
    file_: UploadFile
) -> Response:
    file_path = file_.save_sync("files")

    ProcessFileUpload(
        ProcessFileUploadPayload(file_path=file_path)
    ).dispatch_sync()

    ...

Delaying jobs

By default, jobs are dispatched immediately, but you can also delay them to be executed at a later time. To do that you can prepare the job before dispatching it and use the delay method to specify the delay in seconds:

await ProcessFileUpload(
    ProcessFileUploadPayload(file_path=file_path)
).delay(60).dispatch()   # Delay the job for 60 seconds

ProcessFileUpload(
    ProcessFileUploadPayload(file_path=file_path)
).delay(60).dispatch_sync()   # Delay the job for 60 seconds

Alternatively, you can also specify the delay directly in the job definition, either by using the delay() decorator or by setting the delay option in the job options:

from expanse.jobs.decorators import delay


@delay(60)  # Delay the job for 60 seconds
class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    ...

from expanse.types.jobs.job_options import JobOptions


class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    options: JobOptions = {
        "delay": 60  # Delay the job for 60 seconds
    }

Dispatching jobs to specific transports

By default, jobs are dispatched to the default transport of the message bus, but you can also specify a different transport to dispatch the job to. To do that you can prepare the job before dispatching it and use the via method to specify the transport to use:

await ProcessFileUpload(
    ProcessFileUploadPayload(file_path=file_path)
).via("other_transport").dispatch_sync()

ProcessFileUpload(
    ProcessFileUploadPayload(file_path=file_path)
).via("other_transport").dispatch_sync()

Alternatively, you can also specify the transport directly in the job definition, either by using the via() decorator or by setting the transport option in the job options:

from expanse.jobs.decorators import transport


@transport("other_transport")  # Dispatch the job to the "other_transport" transport
class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    ...

from expanse.types.jobs.job_options import JobOptions


class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    options: JobOptions = {
        "transport": "other_transport"  # Dispatch the job to the "other_transport" transport
    }

Ensuring uniqueness

If you want to ensure that a job is unique and not dispatched multiple times, you can use the unique() decorator or by setting the unique option in the job definition. This will prevent the same job from being dispatched if it is already in the queue.

from expanse.jobs.decorators import unique


@unique()
class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    ...

from expanse.types.jobs.job_options import JobOptions


class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    options: JobOptions = {
        "unique": True
    }