Build a Task Management API

Build an async FastAPI backend with JWT auth and PostgreSQL.

Introduction

30 Second Summary

Every app needs a backend that handles users, authentication, and data. Whether it's a to-do list, a project tracker, or a social platform, the core pattern is the same: register users, verify who they are, and let them manage their own data securely.

In this project, you will build a Task Management API using FastAPI, async SQLAlchemy 2.0, and JWT authentication. You'll create a complete backend with user registration, login, and protected CRUD endpoints for tasks.

What You'll Build

A REST API where users can register, log in with JWT tokens, and manage personal tasks through async database operations powered by SQLAlchemy 2.0 and PostgreSQL.

By the end of this project, you'll have:

  • A FastAPI application with a structured project layout including models, schemas, routes, and dependencies.
  • A JWT authentication system with Argon2 password hashing, user registration, and login endpoints.
  • Protected CRUD endpoints for tasks with ownership-based access control and async database operations.
  • Secret Mission: Add task filtering by status and a profile endpoint with task statistics.

Are there any prerequisites?

You should be comfortable with basic Python concepts like type hints, async/await, and decorators. Familiarity with REST APIs (what endpoints, HTTP methods, and JSON request/response bodies are) will also help you follow along.

Configure the Database Layer

Your project files are in place and all dependencies are installed. Now you need to connect your application to a real database.

Every FastAPI application that persists data needs a database layer. This step builds that layer from the ground up: environment configuration, an async database connection, and the data models that define what gets stored.

In this step, get ready to:
  • Create a .env file and a settings module that loads configuration from environment variables.
  • Build an async database connection with proper session lifecycle management.
  • Define User and Task models using SQLAlchemy 2.0 type-annotated syntax.
Create the database and environment configuration

Before your app can talk to PostgreSQL, the database it will connect to needs to exist. The connection URL in your config will point to a database called taskdb.

  • Create the taskdb database by running this command:
createdb -U postgres taskdb

Enter the password you set for the postgres superuser during installation when prompted.

What does this command do?

The createdb utility is a PostgreSQL tool that creates a new database. The -U postgres flag says to connect as the postgres superuser. If the command produces no output, it succeeded.

Seeing 'createdb: command not found'?

PostgreSQL's bin directory may not be in your PATH. On Windows, the EDB installer typically places it at C:\Program Files\PostgreSQL\18\bin. Add that directory to your system PATH, then open a new terminal and try again.

Alternatively, use the full path directly:

"C:\Program Files\PostgreSQL\18\bin\createdb" -U postgres taskdb

Still stuck? Help me fix the createdb command not found error on Windows.

Now you need a .env file to store configuration values that your app will load at startup. This keeps secrets and connection details out of your source code.

  • Create a .env file in the root of your task-api folder with the following content:
DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/taskdb
SECRET_KEY=09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

What do these values mean?

  • DATABASE_URL tells SQLAlchemy to use the asyncpg driver to connect to your local PostgreSQL database.
  • SECRET_KEY is a random hex string used to sign JWT tokens. In production, you would generate your own with openssl rand -hex 32.
  • ALGORITHM sets the JWT signing algorithm to HS256.
  • ACCESS_TOKEN_EXPIRE_MINUTES controls how long a login token stays valid.

Use your own password in DATABASE_URL

If you set a password other than postgres during PostgreSQL installation, replace the second postgres in the URL with your actual password. The format is user:password@host:port/database.

Next, create the settings module that loads these values into your application.

  • Open app/config.py in VS Code and add the following code:
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/taskdb"
    SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    model_config = {"env_file": ".env"}


settings = Settings()

What does this code do?

  • Settings extends BaseSettings from pydantic-settings. This automatically reads matching environment variables and validates their types.
  • The default values act as fallbacks if the .env file is missing, but model_config = {"env_file": ".env"} tells Pydantic to load values from your .env file automatically.
  • The settings instance at the bottom is what the rest of your app imports to access these values.
  • Save app/config.py.
  • Verify the settings load by running this in your terminal:
python -c "from app.config import settings; print(settings.DATABASE_URL)"

You should see your database URL printed in the terminal.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated (you should see (venv) in your terminal prompt). Also confirm you are running the command from the task-api folder, not from inside the app/ subfolder.

Need more help? Help me debug why Python can't find my app.config module.

✔️ Awesome, I've got everything!

Great. Make sure both .env and app/config.py are saved.

ⓧ I'd like to double check the full code

Here is .env:

DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/taskdb
SECRET_KEY=09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

Here is app/config.py:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/taskdb"
    SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    model_config = {"env_file": ".env"}


settings = Settings()
Build the async database connection

With configuration in place, you can now create the database engine and session factory. These are the building blocks that let your FastAPI routes talk to PostgreSQL without blocking the event loop.

  • Open app/database.py and add the following code:
from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.config import settings

# Create the async engine using the asyncpg driver
engine = create_async_engine(settings.DATABASE_URL, echo=False)

# Session factory that produces AsyncSession instances
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    # Dependency that provides a database session to each request
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

What does this code do?

  • create_async_engine creates a connection pool to PostgreSQL using the asyncpg driver. The echo=False setting hides SQL debug output.
  • async_sessionmaker is a factory that produces database sessions. The expire_on_commit=False setting prevents errors when FastAPI accesses model attributes after the session commits.
  • get_db is an async generator that FastAPI uses as a dependency. Each request gets its own session. If the request succeeds, changes are committed. If an exception occurs, changes are rolled back.

Why expire_on_commit=False?

By default, SQLAlchemy expires all attributes on a model after a commit. In async code, accessing those expired attributes triggers a lazy load that fails because the session is no longer active. Setting expire_on_commit=False keeps attributes available after commit so FastAPI can serialize them into JSON responses.

  • Save app/database.py.
  • Verify the module loads by running:
python -c "from app.database import engine, get_db; print('DB config OK')"

You should see DB config OK printed with no errors.

Seeing an import error?

If you see an error about pydantic_settings, confirm you installed pydantic-settings in Step 1. Run pip install pydantic-settings to fix it.

If you see an error about greenlet, install the async extra by running pip install "sqlalchemy[asyncio]" and try again.

Still stuck? Help me fix my database module import error.

✔️ Awesome, I've got everything!

Great. Make sure app/database.py is saved.

ⓧ I'd like to double check the full code

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=False)

async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
Define the database models

Models define the shape of data in your database. You need two tables: one for users and one for tasks. Each task belongs to a user, creating a one-to-many relationship.

  • Open app/models.py and add the following code:
from datetime import datetime

from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    tasks: Mapped[list["Task"]] = relationship(back_populates="owner")


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(255))
    description: Mapped[str | None] = mapped_column(String(1000), default=None)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    owner: Mapped["User"] = relationship(back_populates="tasks")

What does this code do?

Base is the parent class for all models. SQLAlchemy uses it to track which tables to create.

The User model stores login credentials. The email column is unique and indexed for fast lookups. The tasks relationship gives you access to all of a user's tasks.

The Task model stores individual tasks. The owner_id foreign key links each task to a user. The status column defaults to pending and can be set to in_progress or done.

The Mapped type hints and mapped_column are SQLAlchemy 2.0's declarative syntax. They give you full type-checking support in your editor.

  • Save app/models.py.
  • Verify the full database layer loads by running:
python -c "from app.database import engine, get_db; print('DB config OK')"

You should see DB config OK printed again with no errors.

Seeing a syntax error in models.py?

The Mapped[str | None] syntax requires Python 3.10 or higher. Since you have Python 3.14, this should work. Double-check that you copied the relationship lines exactly, including the quotes around "Task" and "User". These forward references must be strings.

Need help? Help me debug a syntax error in my SQLAlchemy models.

✔️ Awesome, I've got everything!

Great. Make sure app/models.py is saved.

ⓧ I'd like to double check the full code

from datetime import datetime

from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    tasks: Mapped[list["Task"]] = relationship(back_populates="owner")


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(255))
    description: Mapped[str | None] = mapped_column(String(1000), default=None)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    owner: Mapped["User"] = relationship(back_populates="tasks")

Your database layer is solid. You have configuration loading from environment variables, an async connection pool, and fully typed data models. Next up, you will build the authentication system that secures your API with JWT tokens.

Build the Authentication System

Your database layer is ready with async sessions and declarative models. Now the API needs a way to identify who is making each request.

In this step, you'll build a complete authentication system using Argon2 password hashing, JWT tokens for stateless sessions, and Pydantic V2 schemas for request/response validation.

In this step, get ready to:
  • Create the auth module with password hashing and JWT token generation.
  • Define Pydantic schemas for users, tokens, and tasks.
  • Build registration and login endpoints that issue JWT tokens.
Set up the auth module with password hashing and JWT

The auth module handles three responsibilities: hashing passwords so they are never stored in plain text, creating JWT tokens after login, and verifying tokens on every protected request.

  • Open app/auth.py in VS Code and add the imports, password hasher, and OAuth2 scheme by pasting this code:
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.database import get_db
from app.models import User
from app.schemas import TokenData

# Argon2 is the recommended algorithm for password hashing
password_hash = PasswordHash.recommended()

# Tells FastAPI where the login endpoint lives for the Swagger UI "Authorize" button
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


def hash_password(password: str) -> str:
    # One-way hash so the original password is never stored
    return password_hash.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    # Compares a plain-text password against its stored hash
    return password_hash.verify(plain_password, hashed_password)

What does this code do?

PasswordHash.recommended() creates a hasher that uses Argon2, the current gold standard for password hashing. It is intentionally slow to compute, which makes brute-force attacks impractical.

OAuth2PasswordBearer(tokenUrl="/auth/token") tells FastAPI where your login endpoint lives. This powers the "Authorize" button in the auto-generated Swagger UI.

  • Below the verify_password function, add the token creation function:
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    # Copy the data so we don't mutate the original
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        # Default to 15 minutes if no expiry is specified
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    # Sign the token with your secret key
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

What does this code do?

This function builds a JWT payload with an expiration timestamp, then signs it with your secret key using the HS256 algorithm. The resulting token string is what clients send in the Authorization header for every protected request.

  • Below create_access_token, add the dependency that extracts and validates the current user from a request's token:
async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # Decode the JWT and extract the user ID from the 'sub' claim
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        user_id = payload.get("sub")
        if user_id is None:
            raise credentials_exception
        token_data = TokenData(user_id=int(user_id))
    except InvalidTokenError:
        raise credentials_exception

    # Look up the user in the database
    result = await db.execute(select(User).where(User.id == token_data.user_id))
    user = result.scalar_one_or_none()
    if user is None:
        raise credentials_exception
    return user

What does this code do?

get_current_user is a FastAPI dependency. Any route that declares it as a parameter automatically requires a valid JWT. If the token is missing, expired, or references a deleted user, FastAPI returns a 401 Unauthorized response before your route code ever runs.

The Annotated[str, Depends(oauth2_scheme)] pattern uses FastAPI's dependency injection to automatically extract the token from the Authorization: Bearer <token> header.

  • Save app/auth.py.

Seeing an import error for TokenData?

The import from app.schemas import TokenData will fail until you create app/schemas.py in the next substep. This is expected.

help me fix an import error in my auth module

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/auth.py before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.database import get_db
from app.models import User
from app.schemas import TokenData

password_hash = PasswordHash.recommended()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


def hash_password(password: str) -> str:
    return password_hash.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)


def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        user_id = payload.get("sub")
        if user_id is None:
            raise credentials_exception
        token_data = TokenData(user_id=int(user_id))
    except InvalidTokenError:
        raise credentials_exception

    result = await db.execute(select(User).where(User.id == token_data.user_id))
    user = result.scalar_one_or_none()
    if user is None:
        raise credentials_exception
    return user
Define the Pydantic schemas

Pydantic schemas define the shape of data flowing in and out of your API. They handle validation, serialization, and automatic documentation in one place.

  • Open app/schemas.py and add the auth-related schemas by pasting this code:
from datetime import datetime

from pydantic import BaseModel, EmailStr


# Schema for user registration requests
class UserCreate(BaseModel):
    email: EmailStr
    password: str


# Schema for user data returned in responses (no password!)
class UserResponse(BaseModel):
    id: int
    email: str
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}


# Schema for the JWT token response after login
class Token(BaseModel):
    access_token: str
    token_type: str


# Internal schema for decoded token data
class TokenData(BaseModel):
    user_id: int | None = None

What does this code do?

UserCreate uses EmailStr which automatically validates that the input is a proper email address. UserResponse deliberately excludes the password field so it never leaks in API responses.

The model_config = {"from_attributes": True} setting tells Pydantic to read data from SQLAlchemy model attributes (like user.email) instead of requiring a dictionary.

  • Below the TokenData class, add the task schemas:
# Schema for creating a new task
class TaskCreate(BaseModel):
    title: str
    description: str | None = None
    status: str = "pending"


# Schema for updating an existing task (all fields optional)
class TaskUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    status: str | None = None


# Schema for task data returned in responses
class TaskResponse(BaseModel):
    id: int
    title: str
    description: str | None
    status: str
    created_at: datetime
    owner_id: int

    model_config = {"from_attributes": True}

What does this code do?

TaskCreate has sensible defaults. If a user does not provide a status, it defaults to "pending". TaskUpdate makes every field optional so the client can send only the fields it wants to change.

TaskResponse includes owner_id so the client always knows which user owns a task.

  • Save app/schemas.py.

You won't see a visible change yet. The schemas are consumed by the route handlers you'll build next.

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/schemas.py before moving on.

ⓧ I'd like to double check the full code

from datetime import datetime

from pydantic import BaseModel, EmailStr


class UserCreate(BaseModel):
    email: EmailStr
    password: str


class UserResponse(BaseModel):
    id: int
    email: str
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    user_id: int | None = None


class TaskCreate(BaseModel):
    title: str
    description: str | None = None
    status: str = "pending"


class TaskUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    status: str | None = None


class TaskResponse(BaseModel):
    id: int
    title: str
    description: str | None
    status: str
    created_at: datetime
    owner_id: int

    model_config = {"from_attributes": True}
Build the registration and login routes

The auth routes give users a way to create an account and receive a JWT token. Every other endpoint in the API will require that token.

  • Open app/routes_auth.py and add the imports, router, and registration endpoint by pasting this code:
from datetime import timedelta
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import create_access_token, hash_password, verify_password
from app.config import settings
from app.database import get_db
from app.models import User
from app.schemas import Token, UserCreate, UserResponse

# All routes in this file get the /auth prefix
router = APIRouter(prefix="/auth", tags=["auth"])


@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user_data: UserCreate, db: Annotated[AsyncSession, Depends(get_db)]):
    # Check if the email is already taken
    result = await db.execute(select(User).where(User.email == user_data.email))
    existing_user = result.scalar_one_or_none()
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")

    # Create the user with a hashed password
    new_user = User(
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    await db.flush()
    await db.refresh(new_user)
    return new_user

What does this code do?

The /register endpoint first checks if the email already exists. If it does, it returns a 400 error. Otherwise, it hashes the password with Argon2 and stores the new user.

db.flush() sends the INSERT to the database without closing the transaction. db.refresh(new_user) reloads the user object so auto-generated fields like id and created_at are populated before returning the response.

  • Below the register function, add the login endpoint:
@router.post("/token", response_model=Token)
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    # Look up the user by email (OAuth2 form uses 'username' field for email)
    result = await db.execute(select(User).where(User.email == form_data.username))
    user = result.scalar_one_or_none()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    # Create a JWT with the user's ID as the 'sub' claim
    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": str(user.id)}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")

What does this code do?

The /token endpoint receives a standard OAuth2 password form (username + password). It verifies the credentials, then mints a signed JWT containing the user's ID in the sub claim.

OAuth2PasswordRequestForm is a FastAPI utility that parses the form fields that OAuth2 clients expect. The username field holds the email address in our case.

  • Save app/routes_auth.py.

Seeing a circular import error?

If Python complains about circular imports between auth.py and schemas.py, make sure both files are saved. The import chain is routes_auth -> auth -> schemas, which only works if all three files exist.

help me debug a circular import error in my FastAPI project

✔️ Awesome, I've got everything!

Great. Make sure you have saved app/routes_auth.py before moving on.

ⓧ I'd like to double check the full code

from datetime import timedelta
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import create_access_token, hash_password, verify_password
from app.config import settings
from app.database import get_db
from app.models import User
from app.schemas import Token, UserCreate, UserResponse

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user_data: UserCreate, db: Annotated[AsyncSession, Depends(get_db)]):
    result = await db.execute(select(User).where(User.email == user_data.email))
    existing_user = result.scalar_one_or_none()
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")

    new_user = User(
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    await db.flush()
    await db.refresh(new_user)
    return new_user


@router.post("/token", response_model=Token)
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    result = await db.execute(select(User).where(User.email == form_data.username))
    user = result.scalar_one_or_none()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": str(user.id)}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")

Now let's wire everything up with a minimal entry point so you can test registration in the browser.

  • Open app/main.py and paste this temporary starter code:
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import engine
from app.models import Base
from app.routes_auth import router as auth_router


# Create database tables on startup
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(title="Task Management API", lifespan=lifespan)

app.include_router(auth_router)

What does this code do?

The lifespan function runs once when the server starts. It uses Base.metadata.create_all to create the users and tasks tables in PostgreSQL if they don't already exist.

You'll expand this file in the next step to include the tasks router and a root endpoint. For now, it's enough to test the auth system.

  • Save app/main.py.
  • Before starting the server, create the taskdb database in PostgreSQL by running this command:
psql -U postgres -c "CREATE DATABASE taskdb;"

What does this command do?

This connects to PostgreSQL as the postgres superuser and creates a new database called taskdb. Your .env file references this database name in the DATABASE_URL.

  • Enter the password you set during PostgreSQL installation when prompted.
  • Start the development server by running:
uvicorn app.main:app --reload
  • Open your browser and go to http://127.0.0.1:8000/docs.

You should see FastAPI's auto-generated Swagger UI with two endpoints listed under the auth section: POST /auth/register and POST /auth/token.

  • Click the POST /auth/register endpoint to expand it.
  • Click **Try it out**.
  • Replace the request body with this JSON:
{
  "email": "test@example.com",
  "password": "secretpass123"
}
  • Click **Execute**.

You should see a 201 response containing the new user's id, email, is_active, and created_at fields. No password is returned.

Seeing a connection error when the server starts?

Make sure PostgreSQL is running and the taskdb database exists. If the CREATE DATABASE command returned an error saying the database already exists, that is fine.

Also check that your .env file has DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/taskdb with the correct password you set during installation.

help me fix a database connection error in my FastAPI project

Your authentication system is fully wired. Next up, you'll build the task CRUD endpoints that use get_current_user to ensure only logged-in users can manage their own tasks.

Build the Task CRUD Endpoints

You have a working authentication system. Users can register and get a JWT token, but there is nothing to protect yet.

In this step, you will build the actual task endpoints that sit behind that authentication wall. Every endpoint will use the get_current_user dependency to enforce that users can only see and modify their own tasks.

In this step, get ready to:
  • Build create and list endpoints for tasks.
  • Add get, update, and delete endpoints with ownership checks.
  • Wire all routers into a FastAPI app with automatic table creation.
Set up the task creation and listing endpoints

The app/routes_tasks.py file is currently empty. You will add a router that handles task creation and retrieval, with every endpoint requiring a valid JWT.

  • Open app/routes_tasks.py in VS Code and paste the following code:
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import get_current_user
from app.database import get_db
from app.models import Task, User
from app.schemas import TaskCreate, TaskResponse, TaskUpdate

router = APIRouter(prefix="/tasks", tags=["tasks"])


@router.post("", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(
    task_data: TaskCreate,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    # Create a new task and assign it to the authenticated user
    new_task = Task(
        title=task_data.title,
        description=task_data.description,
        status=task_data.status,
        owner_id=current_user.id,
    )
    db.add(new_task)
    await db.flush()
    await db.refresh(new_task)
    return new_task


@router.get("", response_model=list[TaskResponse])
async def list_tasks(
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    # Only return tasks belonging to the current user
    result = await db.execute(select(Task).where(Task.owner_id == current_user.id))
    return result.scalars().all()

What does this code do?

The router uses prefix="/tasks" so every endpoint starts with /tasks. The tags=["tasks"] groups these endpoints together in Swagger UI.

Both functions declare current_user: Annotated[User, Depends(get_current_user)] as a parameter. FastAPI calls get_current_user automatically before each request, rejecting unauthenticated callers with a 401.

The create_task function sets owner_id to the authenticated user's ID, ensuring tasks always belong to whoever created them. The list_tasks function filters by owner_id so users only see their own tasks.

  • Save app/routes_tasks.py.

Why dependency injection?

Instead of manually extracting the token and querying the database in every endpoint, Depends(get_current_user) handles all of that in one place. If you ever change how authentication works, you update one function instead of every route.

Add get, update, and delete endpoints

Individual task operations need to verify that the task exists and belongs to the requesting user. Each endpoint queries by both task_id and owner_id to enforce ownership.

  • In app/routes_tasks.py, add the following code below the list_tasks function:
@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(
    task_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    # Fetch task only if it belongs to the current user
    result = await db.execute(
        select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    )
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@router.put("/{task_id}", response_model=TaskResponse)
async def update_task(
    task_id: int,
    task_data: TaskUpdate,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    # Look up the task with ownership check
    result = await db.execute(
        select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    )
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    # Only update fields that were explicitly provided
    if task_data.title is not None:
        task.title = task_data.title
    if task_data.description is not None:
        task.description = task_data.description
    if task_data.status is not None:
        task.status = task_data.status

    await db.flush()
    await db.refresh(task)
    return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
    task_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    # Verify ownership before deleting
    result = await db.execute(
        select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    )
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    await db.delete(task)

What does this code do?

  • get_task fetches a single task by ID, but only if the current user owns it. If the task does not exist or belongs to someone else, it returns a 404.
  • update_task uses the TaskUpdate schema where all fields are optional. It only overwrites fields the caller explicitly sent, leaving the rest unchanged.
  • delete_task removes the task from the database and returns a 204 No Content response with no body.

The ownership check in every endpoint means that even if a user guesses another user's task ID, they get a 404 instead of the data.

  • Save app/routes_tasks.py.

Seeing a red underline on the imports?

If your editor shows an import error for TaskUpdate, make sure the import at the top of the file includes it: from app.schemas import TaskCreate, TaskResponse, TaskUpdate.

Help me check my imports.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/routes_tasks.py before moving on.

ⓧ I'd like to double check the full code

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import get_current_user
from app.database import get_db
from app.models import Task, User
from app.schemas import TaskCreate, TaskResponse, TaskUpdate

router = APIRouter(prefix="/tasks", tags=["tasks"])


@router.post("", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task(
    task_data: TaskCreate,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    new_task = Task(
        title=task_data.title,
        description=task_data.description,
        status=task_data.status,
        owner_id=current_user.id,
    )
    db.add(new_task)
    await db.flush()
    await db.refresh(new_task)
    return new_task


@router.get("", response_model=list[TaskResponse])
async def list_tasks(
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    result = await db.execute(select(Task).where(Task.owner_id == current_user.id))
    return result.scalars().all()


@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(
    task_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    result = await db.execute(
        select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    )
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@router.put("/{task_id}", response_model=TaskResponse)
async def update_task(
    task_id: int,
    task_data: TaskUpdate,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    result = await db.execute(
        select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    )
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    if task_data.title is not None:
        task.title = task_data.title
    if task_data.description is not None:
        task.description = task_data.description
    if task_data.status is not None:
        task.status = task_data.status

    await db.flush()
    await db.refresh(task)
    return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
    task_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    result = await db.execute(
        select(Task).where(Task.id == task_id, Task.owner_id == current_user.id)
    )
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    await db.delete(task)
Wire everything together in main.py

The app/main.py file is still empty. It needs to create the FastAPI application, register both routers, and ensure database tables exist before serving requests.

  • Open app/main.py in VS Code and paste the following code:
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import engine
from app.models import Base
from app.routes_auth import router as auth_router
from app.routes_tasks import router as tasks_router


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create all database tables on startup
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(title="Task Management API", lifespan=lifespan)

app.include_router(auth_router)
app.include_router(tasks_router)


@app.get("/")
async def root():
    return {"message": "Task Management API is running"}

What does this code do?

  • The lifespan function runs once when the server starts. It opens a database connection and calls Base.metadata.create_all to create the users and tasks tables if they do not already exist.
  • include_router(auth_router) and include_router(tasks_router) attach all the endpoints from both router files to the app.
  • The root GET / endpoint is a simple health check that confirms the API is running.
  • Save app/main.py.
  • Switch back to your terminal. Uvicorn should auto-reload and print a startup complete message.
  • Open http://127.0.0.1:8000/docs in your browser. You should see Swagger UI with three endpoint groups: default, auth, and tasks.

Server showing an import error?

Check that both app/routes_tasks.py and app/routes_auth.py are saved. A missing import in either file prevents the app from starting.

If you see a database connection error, confirm PostgreSQL is running and that your .env file has the correct DATABASE_URL.

Help me debug my server startup error.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py before moving on.

ⓧ I'd like to double check the full code

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import engine
from app.models import Base
from app.routes_auth import router as auth_router
from app.routes_tasks import router as tasks_router


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(title="Task Management API", lifespan=lifespan)

app.include_router(auth_router)
app.include_router(tasks_router)


@app.get("/")
async def root():
    return {"message": "Task Management API is running"}

Now test the full flow to confirm everything works together.

  • In Swagger UI, expand POST /auth/register and click Try it out.
  • Enter a test email and password in the request body, then click Execute.
  • Expand POST /auth/token, click Try it out, enter the same email in username and your password in password, then click Execute.
  • Copy the access_token value from the response.
  • Click the Authorize button at the top of the page.
  • Paste your token into the Value field and click Authorize, then Close.
  • Expand POST /tasks, click Try it out, and create a task with a title like Fix login bug. Click Execute.
  • Expand GET /tasks and click Execute. You should see your task in the response array.
  • Test PUT /tasks/{task_id} to update the status to done, then DELETE /tasks/{task_id} to remove it.

How does ownership enforcement work?

Every task query includes Task.owner_id == current_user.id in the WHERE clause. If User A tries to access a task belonging to User B, the query returns no rows and the endpoint responds with a 404.

This pattern is simpler and more secure than fetching the task first and then checking ownership separately. The database does the filtering in a single query.

Getting a 401 Unauthorized on task endpoints?

Make sure you clicked Authorize and pasted the full token value (without quotes). The token expires after 30 minutes, so if you waited too long, request a new one from POST /auth/token.

Help me figure out why my task endpoints return 401.

Your API now has a complete set of authenticated CRUD endpoints. Next up, you will tackle the Secret Mission and add filtering and user statistics to make the API even more powerful.

Secret mission

Add Filtering and User Stats

Your API works, but real users need to find specific tasks fast. Can you add query parameter filtering to the task list and build a profile endpoint that shows task statistics?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The taskdb PostgreSQL database.
  • The task-api project directory (includes virtual environment and all code files).
  • PostgreSQL 18 (installed during this project).
  • The Uvicorn development server (running process).

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with the API.

  • Your Uvicorn server continues running in the terminal where you started it.
  • PostgreSQL continues running as a background service on port 5432.
  • Your task-api folder stays on your Desktop with all code intact.

Pause - I'll come back to this later

Shut down running processes to free up system resources while keeping your files and database for later.

  • In the terminal running Uvicorn, press Ctrl+C to stop the development server.
  • Deactivate your virtual environment by running:
deactivate
  • Stop the PostgreSQL service to free memory. Open the Windows Services app (press the Windows key and search for Services), find postgresql-x64-18, right-click it, and select Stop.
  • To restart later, right-click the same service and select Start, then reactivate your virtual environment with venv\Scripts\activate from the task-api folder.

Delete - I don't want to use this again

Remove all project resources and start fresh. This is irreversible.

  • In the terminal running Uvicorn, press Ctrl+C to stop the server.
  • Deactivate your virtual environment by running:
deactivate
  • Drop the taskdb database by running:
psql -U postgres -c "DROP DATABASE taskdb;"
  • Delete the task-api project folder. Navigate to your Desktop and run:
cd ~/Desktop
rmdir /s /q task-api

macOS or Linux

Use rm -rf task-api instead of rmdir /s /q.

  • To uninstall PostgreSQL 18, open the Windows Control Panel and go to Uninstall or Change a Program.
  • Right-click PostgreSQL 18 in the program list and select Uninstall/Change.
  • Select Entire application and click Next to remove PostgreSQL completely.

Nice Work!

Nice Work!

You just built a production-grade Task Management API from scratch using FastAPI, async SQLAlchemy 2.0, and JWT authentication.

You've learned how to:

  • Build a FastAPI application with async SQLAlchemy 2.0 connected to PostgreSQL via the asyncpg driver for non-blocking database access.
  • Implement JWT authentication with Argon2 password hashing and OAuth2 password flow following FastAPI's official patterns.
  • Create protected CRUD endpoints with ownership-based access control using FastAPI's dependency injection for both database sessions and user authentication.
  • Secret Mission: Added dynamic query filtering by status and search terms, plus a user profile endpoint with task statistics.

Ready to quiz yourself?