Prefect Background Jobs in FastAPI

Posted: 2026-07-26

Four years ago I wrote about Better FastAPI Background Jobs, where I’d started using APScheduler for my home automation projects. I liked that it ran jobs from a variety of triggers in the same runtime as the API which kept the runtime environment simple.

I use Prefect at work, and appreciate the observability and mature orchestration features that it makes easily accessible. I still wanted to keep the simplicity of the in-app runtime though, so I set out to recreate a similar setup as I had with APScheduler.

Prefect have a post that details setting up simple background tasks for a web app, with a relatively complex worker setup, but I wanted full workflows aka flows

This post details the setup, along with a surprisingly deep hole I fell into working out why it was messing up my existing uvicorn logs.

Naive Implementation

One might expect that this should be fairly straightforward, like so

example.py:

import logging

import prefect
from fastapi import FastAPI
from pydantic import BaseModel

logger = logging.getLogger(__name__)


@prefect.task
def hello_world():
    run_logger = prefect.get_run_logger()
    run_logger.info('Hello World')


@prefect.flow
def demo():
    hello_world()


app = FastAPI()


class Message(BaseModel):
    message: str


@app.get('/', response_model=Message)
async def root():
    logger.info('Test')
    return {'message': 'Hello World'}

However, if you use this with a normal uvicorn setup, it changes the log format and hides some logs

2026-07-26 23:32:45,787 - example - INFO - Test
2026-07-26 23:32:45,792 - example - INFO - Test

The default uvicorn logging setup that I was used to seeing was getting clobbered by prefect’s logging format

Powerful tools can have sharp edges, or unexpected side effects!

Why it happens

Accessing attributes of the prefect module (including from prefect import flow, task) calls logging.config.dictConfig with Prefect’s own config.

When dictConfig configures a logger, any existing children of that logger get reset

Prefect configures uvicorn, so uvicorn.error and uvicorn.access get reset

Resolving the logging

example.py:

import logging
import logging.config
from pathlib import Path

import prefect
import yaml
from fastapi import FastAPI
from pydantic import BaseModel


def reset_logging():
    """Fix prefect overriding the logging"""
    # trigger prefect's lazy logging config side effect
    import prefect.main  # noqa: F401, PLC0415

    logging.config.dictConfig(
        yaml.safe_load((Path(__file__).parent / 'logging.yaml').read_text()),
    )


reset_logging()
logger = logging.getLogger(__name__)


@prefect.task
def hello_world():
    run_logger = prefect.get_run_logger()
    run_logger.info('Hello World')


@prefect.flow
def demo():
    hello_world()


app = FastAPI()


class Message(BaseModel):
    message: str


@app.get('/', response_model=Message)
async def root():
    logger.info('Test')
    return {'message': 'Hello World'}

with logging.yaml:

---
version: 1
disable_existing_loggers: false

formatters:
  default:
    format: '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
handlers:
  default:
    formatter: default
    class: logging.StreamHandler
    stream: ext://sys.stdout
loggers:
  example:  # Match to app logger name (usually __name__ of entrypoint file)
    level: INFO
  uvicorn.error:
    level: INFO
    handlers: [default]
    propagate: false
  uvicorn.access:
    level: INFO
    handlers: [default]
    propagate: false
root:
  # WARNING keeps noisy third-party libraries quiet
  level: WARNING
  handlers: [default]

Full setup

example.py:

import asyncio
import logging
import logging.config
from contextlib import asynccontextmanager
from datetime import datetime, timedelta
from pathlib import Path
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from collections.abc import Coroutine

import prefect
import yaml
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI
from pydantic import BaseModel


def reset_logging():
    """Fix prefect overriding the logging"""
    # trigger prefect's lazy logging config side effect
    import prefect.main  # noqa: F401, PLC0415

    logging.config.dictConfig(
        yaml.safe_load((Path(__file__).parent / 'logging.yaml').read_text()),
    )


reset_logging()
logger = logging.getLogger(__name__)


@prefect.task
def hello_world():
    run_logger = prefect.get_run_logger()
    run_logger.info('Hello World')


@prefect.flow
def demo():
    hello_world()


scheduler = AsyncIOScheduler()


# APScheduler scheduled jobs are duplicated per worker
@scheduler.scheduled_job('interval', minutes=1)
async def example_heartbeat():
    logger.info(f'Time: {datetime.now()}')


_background_tasks = set()


def _on_background_task_done(task: asyncio.Task):
    _background_tasks.discard(task)
    # An unretrieved task exception is otherwise only reported when the task is
    # garbage collected, which here means a crashed `aserve` goes unnoticed: the
    # API carries on answering requests while every scheduled flow run has
    # quietly stopped.
    if not task.cancelled() and (exc := task.exception()) is not None:
        logger.error('Background task failed', exc_info=exc)


def background_task(coro: Coroutine[Any, Any, None]):
    # Keep a strong reference to the task so it isn't garbage collected mid-run,
    # then drop it once it completes.
    # https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
    task = asyncio.create_task(coro)
    _background_tasks.add(task)
    task.add_done_callback(_on_background_task_done)


async def startup_scheduler():
    scheduler.start()
    logger.info('Finished startup scheduler')


async def startup_prefect():
    # Prefect flow deployments overwrite per worker, and only one worker
    # will be given the run by the server
    hello_deploy = await demo.to_deployment(
        'hello world',
        tags=['demo'],
        interval=timedelta(days=1),
    )
    background_task(
        prefect.aserve(
            hello_deploy,
            limit=2,
            print_starting_message=False,  # reduce log noise
        ),
    )
    logger.info('Finished startup prefect')


@asynccontextmanager
async def lifespan(_app: FastAPI):
    await asyncio.gather(
        startup_prefect(),
        startup_scheduler(),
    )
    logger.info('Finished startup in lifespan')
    yield
    scheduler.shutdown()
    # Cancel `aserve` so pause_on_shutdown runs
    tasks = list(_background_tasks)
    for task in tasks:
        task.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    logger.info('Finished shutdown in lifespan')


app = FastAPI(lifespan=lifespan)


class Message(BaseModel):
    message: str


@app.get('/', response_model=Message)
async def root():
    logger.info('Test')
    return {'message': 'Hello World'}

docker-compose.yml brings up the app next to a Prefect server, so there is a UI at http://localhost:4200 to look at flow runs in:

---
services:
  web:
    build: .
    # --log-config is deliberately redundant: uvicorn applies it before it
    # imports the app, and Prefect's setup_logging() then replaces the whole
    # config, so the flag alone achieves nothing. example.py re-applies the same
    # file after that, and that is the pass that survives. The flag stays
    # because the obvious-looking approach failing is the point worth seeing.
    # --reload is for local development only; drop it (and the bind mount below)
    # if you adapt this for anything else.
    command: >
      uvicorn example:app --host 0.0.0.0 --port 8000
      --log-config logging.yaml --reload
    ports:
      # Loopback only. Nothing here is authenticated, so publishing on every
      # host interface would put it in front of anyone on the same network.
      - "127.0.0.1:8000:8000"
    volumes:
      # Serves the host's code instead of the copy baked into the image, so
      # --reload above picks up edits without a rebuild.
      - .:/code
    environment:
      # Reach the Prefect server over the compose network by service name
      PREFECT_API_URL: http://prefect-server:4200/api
    restart: unless-stopped
    depends_on:
      prefect-server:
        condition: service_healthy


  prefect-server:
    image: prefecthq/prefect:3-latest
    command: prefect server start
    restart: unless-stopped
    ports:
      # Loopback only: the Prefect API is unauthenticated and will happily
      # create flow runs for anyone who can reach it. The UI at
      # http://localhost:4200 still works, and the web container reaches the
      # server over the compose network rather than through this mapping.
      - "127.0.0.1:4200:4200"
    volumes:
      - prefect-data:/root/.prefect  # Persist Prefect DB
    environment:
      # Allow connections from other containers
      PREFECT_SERVER_API_HOST: 0.0.0.0
      # The browser loads the UI from the host, so it must reach the API at a
      # host-reachable address (the published 4200 port), not the 0.0.0.0 bind
      # address the UI would otherwise advertise.
      PREFECT_UI_API_URL: http://localhost:4200/api
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:4200/api/health')"]
      interval: 10s
      timeout: 5s
      retries: 12
      start_period: 20s

volumes:
  # Named volumes for data persistence
  prefect-data: {}

logging.yaml:

---
version: 1
disable_existing_loggers: false

formatters:
  default:
    format: '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
handlers:
  default:
    formatter: default
    class: logging.StreamHandler
    stream: ext://sys.stdout
loggers:
  example:  # Match to app logger name (usually __name__ of entrypoint file)
    level: INFO
  uvicorn.error:
    level: INFO
    handlers: [default]
    propagate: false
  uvicorn.access:
    level: INFO
    handlers: [default]
    propagate: false
root:
  # WARNING keeps noisy third-party libraries quiet
  level: WARNING
  handlers: [default]

pyproject.toml:

[project]
name = "example"
version = "0.1"
dependencies = [
    "apscheduler>=3.11.0",
    "fastapi>=0.116.1",
    "prefect>=3.4.10",
    "pydantic>=2.11.7",
    "pyyaml>=6.0.2",
    "uvicorn[standard]>=0.35.0",
]
requires-python = "==3.14.*"

[tool.uv]
environments = [
    "sys_platform == 'linux' and implementation_name == 'cpython'"
]

[tool.ruff]
target-version = "py314"

[tool.ruff.lint]
select = ["ALL"]
ignore = [
"Q000",
"Q003",
"D", # Docstrings
"ANN", # Annontations
"G004", # fstring in logs
"S311", # random not crypo
"DTZ", # timezone
"TRY003", # errors
"COM812", # conflicts with ruff-format
"CPY001", # copyright notice
]

# Allow fix for all enabled rules (when `--fix`) is provided.
fixable = ["ALL"]
unfixable = []

[tool.ruff.format]
quote-style = "single"

Dockerfile:

ARG UV_VERSION="0.11.28"
ARG PYTHON_VERSION="3.14"
ARG DEBIAN_VERSION="trixie"

FROM ghcr.io/astral-sh/uv:${UV_VERSION} AS uv
FROM python:${PYTHON_VERSION}-slim-${DEBIAN_VERSION} AS builder

COPY --from=uv /uv /uvx /bin/
ENV UV_PROJECT_ENVIRONMENT=/opt/venv

# pyproject.toml has no [build-system], so uv treats this as a non-packaged
# project and there is nothing to install beyond the dependencies: one sync
# does the whole job. Bind-mounting the manifests rather than COPYing them
# keeps the layer keyed on the lockfile alone, so editing example.py doesn't
# reinstall anything.
RUN --mount=type=cache,target=/root/.cache/uv \
    --mount=type=bind,source=uv.lock,target=uv.lock \
    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
    uv sync --locked --no-install-project --compile-bytecode

FROM python:${PYTHON_VERSION}-slim-${DEBIAN_VERSION}

WORKDIR /code

# Only the built virtualenv crosses over, so uv itself never ships.
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

COPY ./example.py ./logging.yaml ./

EXPOSE 8000

CMD ["uvicorn", "example:app", "--host", "0.0.0.0", "--port", "8000", \
     "--log-config", "logging.yaml"]

Tradeoffs with APScheduler

Pros

• Scaling to multiple workers is handled without duplicating scheduled tasks
• Clearer maturity path
  • Separate out the execution of the flows to separate serveing or workers

Cons

• Quirky side effects