Getting Started

Welcome to the FastAPI Task Manager Getting Started Guide! This guide will walk you through the setup process step by step.

---

What you will learn

1. Installation - Install the package in your FastAPI project
2. Configurations - Understand and customize all configuration options
3. API Reference - Explore the built-in management API endpoints
4. Architecture - Understand how the system works under the hood

---

Prerequisites

Before you begin, make sure you have:

• A FastAPI project set up (FastAPI installation guide)
• A running Redis server (Redis installation guide)
• Python 3.10+

---

Quick Overview

FastAPI Task Manager works through three main concepts:

1. TaskManager

The entry point that connects to your FastAPI application and Redis. It handles the lifecycle (start/stop) automatically via FastAPI's lifespan.

from fastapi_task_manager import TaskManager, Config

task_manager = TaskManager(
    app=app,
    config=Config(redis_host="localhost"),
)

2. TaskGroup

Organizes related tasks together. You can have multiple task groups in your application.

from fastapi_task_manager import TaskGroup

my_tasks = TaskGroup(name="My Tasks", tags=["example"])
task_manager.add_task_group(my_tasks)

3. Tasks

Individual scheduled functions, defined with a cron expression via the @add_task decorator.

@my_tasks.add_task("*/5 * * * *", name="cleanup", description="Clean old records")
async def cleanup_task():
    # Your task logic here
    pass

---

Minimal Working Example

Here is a complete, minimal example that you can run immediately:

Python 3.10+

from fastapi import APIRouter, FastAPI
from pydantic_settings import BaseSettings, SettingsConfigDict

from fastapi_task_manager import Config as ManagerConfig  # renamed to avoid conflict with our app Config class
from fastapi_task_manager import TaskGroup, TaskManager


class Config(BaseSettings):
    model_config = SettingsConfigDict(
        # `.env.prod` takes priority over `.env`
        env_file=(".env", ".env.prod"),
        extra="forbid",
    )

    # --------- Redis config variables ---------
    redis_host: str
    redis_port: int = 6379
    redis_password: str | None = None
    redis_db: int = 0
    # --------- End of redis config variables ---------

    # --------- App config variables ---------
    app_name: str = "my_fastapi_app"
    concurrent_tasks: int = 3
    # --------- End of app config variables ---------


CONFIG = Config()  # ty: ignore[missing-argument]

app = FastAPI()
router = APIRouter()

task_manager = TaskManager(
    config=ManagerConfig(
        redis_host=CONFIG.redis_host,
        redis_port=CONFIG.redis_port,
        redis_password=CONFIG.redis_password,
        redis_db=CONFIG.redis_db,
        concurrent_tasks=CONFIG.concurrent_tasks,
        redis_key_prefix=CONFIG.app_name,
    ),
    app=app,
)
my_example_task_group = TaskGroup(
    tags=["example"],
    name="My Example Task Group",
)
task_manager.add_task_group(my_example_task_group)

manager_router = task_manager.get_manager_router()
router.include_router(
    manager_router,
    prefix="/task-manager",
    tags=["task-manager"],
)

app.include_router(router)


@my_example_task_group.add_task(
    "*/5 * * * *",  # Run every 5 minutes
    name="my_scheduled_task",
    description="This is my scheduled task",
)
async def my_scheduled_task():
    pass
    # Your task logic here

This sets up:

• A FastAPI app with the task manager integrated
• A task group with one scheduled task running every 5 minutes
• The management API router under /task-manager/ for monitoring and control

---

Next Steps

• Configure: See Configurations for all available settings including retry backoff, reconciliation, and streams
• Manage tasks at runtime: Learn about Dynamic Tasks to create and delete tasks via the API
• Monitor: Use the built-in API endpoints to monitor health, list tasks, and control execution
• Understand the internals: Read the Architecture guide to learn about leader election, Redis Streams, and fault tolerance