Master REST APIs with FastAPI and PostgreSQL: End-to-end tutorial with source code

Sep 15, 2025

In today’s backend development ecosystem, FastAPI has positioned itself as one of the most popular choices for creating modern and efficient REST APIs. It combines Python’s simplicity with exceptional performance, automatic documentation, and static typing that significantly improves the development experience.

Unlike traditional frameworks like Flask or Django REST, FastAPI leverages modern Python features such as type hints and asynchronous programming to offer better performance, development-time error detection, and automatic documentation with Swagger UI. This translates to fewer bugs, faster development, and more robust APIs.

In this tutorial, we’ll build a complete task management application (Task Manager) that includes user registration, JWT authentication, CRUD operations, and production deployment. The project will be fully functional and serve as a solid foundation for more complex APIs.

We’ll use a modern and proven stack: FastAPI as the main framework, PostgreSQL as the database, SQLAlchemy as ORM, Docker for containerization, and pytest for testing.

Why FastAPI for Modern APIs?

FastAPI has rapidly gained popularity in the Python community for several fundamental reasons that distinguish it from other frameworks:

Exceptional Performance

FastAPI is built on Starlette for request handling and Pydantic for data validation, allowing it to achieve performance comparable to Go or Node.js frameworks. In independent benchmarks, FastAPI consistently outperforms Django REST Framework and approaches FastHTTP performance in Go.

Automatic Documentation

One of the most appreciated features is the automatic generation of interactive documentation. FastAPI automatically creates Swagger UI and ReDoc documentation based on Python type hints, eliminating the need to maintain separate documentation.

Modern Development with Type Hints

Extensive use of static typing allows IDEs like PyCharm or VS Code to offer intelligent autocompletion, real-time error detection, and safe refactoring. This significantly reduces bugs and improves team productivity.

Native Asynchronous Compatibility

FastAPI supports both synchronous and asynchronous functions transparently, allowing you to take full advantage of asynchronous programming when needed without complicating the code.

FastAPI’s superior performance comes from its ASGI-based architecture and the use of optimized libraries like Starlette and Pydantic, resulting in a more agile development experience and faster applications.

Setting Up the Development Environment

Before starting with the code, we need to configure a consistent and reproducible development environment. We’ll use Docker to ensure the environment works the same on all systems.

Project Structure

Code organization is fundamental to maintaining a scalable and maintainable project. FastAPI doesn’t impose a specific structure, but we’ll follow community best practices that clearly separate responsibilities:

We’ll create an organized structure that follows best practices for FastAPI projects:


    mkdir task-manager-api
    cd task-manager-api

    # Directory structure
    mkdir -p {app/{api,core,db,models,schemas,services},tests,scripts}

    # Configuration files
    touch {requirements.txt,Dockerfile,docker-compose.yml,.env.example,.gitignore}

The final structure will look like this:

task-manager-api/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application entry point
│   ├── api/                 # Endpoints and routes
│   │   ├── __init__.py
│   │   ├── deps.py          # Shared dependencies
│   │   └── v1/              # API version 1
│   │       ├── __init__.py
│   │       ├── auth.py      # Authentication endpoints
│   │       └── tasks.py     # Task endpoints
│   ├── core/                # Configuration and utilities
│   │   ├── __init__.py
│   │   ├── config.py        # Environment variables
│   │   └── security.py      # JWT and hashing
│   ├── db/                  # Database
│   │   ├── __init__.py
│   │   ├── database.py      # DB connection
│   │   └── base.py          # Base for models
│   ├── models/              # SQLAlchemy models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── task.py
│   ├── schemas/             # Pydantic models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── task.py
│   └── services/            # Business logic
│       ├── __init__.py
│       ├── auth_service.py
│       └── task_service.py
├── tests/                   # Automated tests
├── scripts/                 # Utility scripts
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
└── .env.example

Project Dependencies

It’s important to understand what each dependency does in our project. Each library has a specific purpose:

FastAPI: The main framework for creating the API
Uvicorn: ASGI server for running FastAPI in production
Pydantic: Data validation and serialization with type hints
SQLAlchemy: ORM for interacting with PostgreSQL in a pythonic way
psycopg2-binary: PostgreSQL driver for Python
python-jose: Library for handling JWT tokens
passlib: Secure password hashing with bcrypt
pytest: Testing framework with excellent FastAPI support

We’ll define the necessary dependencies in requirements.txt:


    # requirements.txt
    fastapi==0.104.1
    uvicorn[standard]==0.24.0
    pydantic==2.5.0
    pydantic-settings==2.1.0
    sqlalchemy==2.0.23
    psycopg2-binary==2.9.9
    alembic==1.13.1
    python-jose[cryptography]==3.3.0
    python-multipart==0.0.6
    passlib[bcrypt]==1.7.4
    pytest==7.4.3
    pytest-asyncio==0.21.1
    httpx==0.25.2

Docker Configuration

Docker will allow us to create a consistent and reproducible environment. The Dockerfile we’ll create is optimized for development, with automatic reloading of changes and mounted volumes to facilitate debugging:

We’ll create a Dockerfile optimized for development:


    # Dockerfile
    FROM python:3.11-slim

    WORKDIR /app

    # Install system dependencies
    RUN apt-get update && apt-get install -y \
        gcc \
        && rm -rf /var/lib/apt/lists/*

    # Copy requirements and install Python dependencies
    COPY requirements.txt .
    RUN pip install --no-cache-dir -r requirements.txt

    # Copy application code
    COPY ./app ./app

    # Expose port
    EXPOSE 8000

    # Default command
    CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

And the docker-compose.yml file for local development:


    # docker-compose.yml
    version: '3.8'

    services:
    api:
        build: .
        ports:
        - "8000:8000"
        volumes:
        - ./app:/app/app
        - ./tests:/app/tests
        environment:
        - DATABASE_URL=postgresql://postgres:postgres@db:5432/taskmanager
        - SECRET_KEY=dev-secret-key-change-in-production
        depends_on:
        - db
        command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

    db:
        image: postgres:15-alpine
        environment:
        - POSTGRES_USER=postgres
        - POSTGRES_PASSWORD=postgres
        - POSTGRES_DB=taskmanager
        ports:
        - "5432:5432"
        volumes:
        - postgres_data:/var/lib/postgresql/data

    volumes:
    postgres_data:

Configuration and Data Models

Robust configuration is the foundation of any application. We’ll use Pydantic Settings which gives us automatic validation of environment variables, type safety, and clear error handling when configuration is missing.

Configuration System

Pydantic Settings allows us to:

Automatic type validation in environment variables
Safe default values
Automatic documentation of required configuration
Perfect integration with Python’s type system

We’ll use Pydantic Settings to handle configuration in a type-safe way:


    # app/core/config.py
    from pydantic_settings import BaseSettings
    from pydantic import PostgresDsn

    class Settings(BaseSettings):
        # Application configuration
        PROJECT_NAME: str = "Task Manager API"
        PROJECT_VERSION: str = "1.0.0"
        API_V1_STR: str = "/api/v1"
        
        # Security configuration
        SECRET_KEY: str
        ALGORITHM: str = "HS256"
        ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
        
        # Database configuration
        DATABASE_URL: PostgresDsn
        
        # CORS configuration
        BACKEND_CORS_ORIGINS: list[str] = ["http://localhost:3000", "http://localhost:8080"]
        
        class Config:
            env_file = ".env"
            case_sensitive = True

    settings = Settings()

Database Connection

SQLAlchemy is Python’s most mature and powerful ORM. While FastAPI supports asynchronous ORMs like SQLModel, traditional SQLAlchemy remains an excellent choice for its stability and ecosystem.

The configuration we’ll implement uses:

Engine: Manages database connections
SessionLocal: Factory for creating database sessions
Base: Base class for all our models
get_db(): Dependency that provides DB sessions to our endpoints

We’ll configure SQLAlchemy:


    # app/db/database.py
    from sqlalchemy import create_engine
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    from app.core.config import settings

    # Create SQLAlchemy engine
    engine = create_engine(str(settings.DATABASE_URL))

    # Create local session
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    # Base for models
    Base = declarative_base()

    # Dependency to get DB session
    def get_db():
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()

Data Models

SQLAlchemy models represent our database tables. Each model defines:

Table structure: Columns, data types, constraints
Relationships: How tables connect to each other
Indexes: To optimize frequent queries
Metadata: Additional information like automatic timestamps

Our User model includes essential fields for authentication and auditing, while Task implements a complete management system with priorities, due dates, and states.

We’ll define SQLAlchemy models for users and tasks:


    # app/models/user.py
    from sqlalchemy import Column, Integer, String, Boolean, DateTime
    from sqlalchemy.orm import relationship
    from sqlalchemy.sql import func
    from app.db.database import Base

    class User(Base):
        __tablename__ = "users"
        
        id = Column(Integer, primary_key=True, index=True)
        email = Column(String, unique=True, index=True, nullable=False)
        username = Column(String, unique=True, index=True, nullable=False)
        hashed_password = Column(String, nullable=False)
        is_active = Column(Boolean, default=True)
        created_at = Column(DateTime(timezone=True), server_default=func.now())
        updated_at = Column(DateTime(timezone=True), onupdate=func.now())
        
        # Relationship with tasks
        tasks = relationship("Task", back_populates="owner", cascade="all, delete-orphan")


    # app/models/task.py
    from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey, Text
    from sqlalchemy.orm import relationship
    from sqlalchemy.sql import func
    from app.db.database import Base

    class Task(Base):
        __tablename__ = "tasks"
        
        id = Column(Integer, primary_key=True, index=True)
        title = Column(String, nullable=False, index=True)
        description = Column(Text, nullable=True)
        is_completed = Column(Boolean, default=False)
        priority = Column(String, default="medium")  # low, medium, high
        due_date = Column(DateTime(timezone=True), nullable=True)
        created_at = Column(DateTime(timezone=True), server_default=func.now())
        updated_at = Column(DateTime(timezone=True), onupdate=func.now())
        
        # Foreign key to user
        owner_id = Column(Integer, ForeignKey("users.id"), nullable=False)
        
        # Relationship with user
        owner = relationship("User", back_populates="tasks")

Pydantic Schemas

Pydantic schemas are the heart of validation in FastAPI. They separate presentation logic from persistence logic:

Input schemas: Validate data coming from the client
Output schemas: Control what data is returned to the client
Update schemas: Allow partial updates with optional fields

This separation gives us:

Security: We never expose sensitive fields like passwords
Flexibility: Different views of the same data depending on context
Validation: Automatic in requests and responses
Documentation: Swagger UI is generated automatically

Pydantic schemas define the data structure for requests and responses:


    # app/schemas/user.py
    from pydantic import BaseModel, EmailStr
    from datetime import datetime
    from typing import Optional

    # Base schema for user
    class UserBase(BaseModel):
        email: EmailStr
        username: str

    # Schema for creating user
    class UserCreate(UserBase):
        password: str

    # Schema for updating user
    class UserUpdate(BaseModel):
        email: Optional[EmailStr] = None
        username: Optional[str] = None
        password: Optional[str] = None

    # Schema for responses (without password)
    class User(UserBase):
        id: int
        is_active: bool
        created_at: datetime
        
        class Config:
            from_attributes = True

    # Schema for login
    class UserLogin(BaseModel):
        username: str
        password: str

    # Schema for token
    class Token(BaseModel):
        access_token: str
        token_type: str = "bearer"
        
    class TokenData(BaseModel):
        username: Optional[str] = None


    # app/schemas/task.py
    from pydantic import BaseModel
    from datetime import datetime
    from typing import Optional

    # Base schema for task
    class TaskBase(BaseModel):
        title: str
        description: Optional[str] = None
        priority: str = "medium"
        due_date: Optional[datetime] = None

    # Schema for creating task
    class TaskCreate(TaskBase):
        pass

    # Schema for updating task
    class TaskUpdate(BaseModel):
        title: Optional[str] = None
        description: Optional[str] = None
        is_completed: Optional[bool] = None
        priority: Optional[str] = None
        due_date: Optional[datetime] = None

    # Schema for responses
    class Task(TaskBase):
        id: int
        is_completed: bool
        created_at: datetime
        updated_at: Optional[datetime] = None
        owner_id: int
        
        class Config:
            from_attributes = True

    # Schema for tasks with owner information
    class TaskWithOwner(Task):
        owner: "User"

Implementing JWT Authentication

Authentication is one of the most critical aspects of any API. JWT (JSON Web Tokens) is the de facto standard for modern APIs because:

Stateless: Doesn’t require storing sessions on the server
Scalable: Perfect for microservices and distributed architectures
Secure: Cryptographically signed to prevent tampering
Flexible: Can include custom claims according to needs

Our system will implement:

Secure password hashing with bcrypt
JWT token generation and verification
Authentication middleware to protect endpoints
Token expiration and renewal handling

We’ll implement a robust system using JWT tokens:

Security System

This module centralizes all cryptographic logic. bcrypt is considered the gold standard for password hashing due to its resistance to brute force attacks and rainbow tables. JWT tokens include an expiration timestamp to limit exposure window in case of compromise.


    # app/core/security.py
    from datetime import datetime, timedelta
    from typing import Optional
    from jose import JWTError, jwt
    from passlib.context import CryptContext
    from app.core.config import settings

    # Configuration for password hashing
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

    def verify_password(plain_password: str, hashed_password: str) -> bool:
        """Verify password against hash"""
        return pwd_context.verify(plain_password, hashed_password)

    def get_password_hash(password: str) -> str:
        """Generate password hash"""
        return pwd_context.hash(password)

    def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
        """Create JWT token"""
        to_encode = data.copy()
        
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
        
        to_encode.update({"exp": expire})
        encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
        return encoded_jwt

    def verify_token(token: str) -> Optional[str]:
        """Verify and extract username from token"""
        try:
            payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
            username: str = payload.get("sub")
            if username is None:
                return None
            return username
        except JWTError:
            return None

Authentication Service

The Service Layer pattern separates business logic from controllers (endpoints). This service encapsulates all user-related operations:

Credential validation
Secure user creation
Duplicate verification
User lifecycle management

This facilitates testing, reusability, and code maintenance.


    # app/services/auth_service.py
    from sqlalchemy.orm import Session
    from app.models.user import User
    from app.schemas.user import UserCreate
    from app.core.security import verify_password, get_password_hash
    from typing import Optional

    class AuthService:
        
        @staticmethod
        def get_user_by_username(db: Session, username: str) -> Optional[User]:
            """Get user by username"""
            return db.query(User).filter(User.username == username).first()
        
        @staticmethod
        def get_user_by_email(db: Session, email: str) -> Optional[User]:
            """Get user by email"""
            return db.query(User).filter(User.email == email).first()
        
        @staticmethod
        def authenticate_user(db: Session, username: str, password: str) -> Optional[User]:
            """Authenticate user with username and password"""
            user = AuthService.get_user_by_username(db, username)
            if not user:
                return None
            if not verify_password(password, user.hashed_password):
                return None
            return user
        
        @staticmethod
        def create_user(db: Session, user: UserCreate) -> User:
            """Create new user"""
            hashed_password = get_password_hash(user.password)
            db_user = User(
                username=user.username,
                email=user.email,
                hashed_password=hashed_password
            )
            db.add(db_user)
            db.commit()
            db.refresh(db_user)
            return db_user
        
        @staticmethod
        def is_username_taken(db: Session, username: str) -> bool:
            """Check if username already exists"""
            return db.query(User).filter(User.username == username).first() is not None
        
        @staticmethod
        def is_email_taken(db: Session, email: str) -> bool:
            """Check if email already exists"""
            return db.query(User).filter(User.email == email).first() is not None

Authentication Dependencies

FastAPI uses a very powerful dependency injection system. These functions:

Run automatically before endpoints
Can be reused across multiple endpoints
Handle token extraction and validation
Provide authenticated users to our endpoints

HTTPBearer automatically handles token extraction from the Authorization header.


    # app/api/deps.py
    from fastapi import Depends, HTTPException, status
    from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
    from sqlalchemy.orm import Session
    from app.db.database import get_db
    from app.core.security import verify_token
    from app.services.auth_service import AuthService
    from app.models.user import User

    # Configure security scheme
    security = HTTPBearer()

    async def get_current_user(
        credentials: HTTPAuthorizationCredentials = Depends(security),
        db: Session = Depends(get_db)
    ) -> User:
        """Get current user from JWT token"""
        
        credentials_exception = HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
        
        # Verify token
        username = verify_token(credentials.credentials)
        if username is None:
            raise credentials_exception
        
        # Get user from DB
        user = AuthService.get_user_by_username(db, username)
        if user is None:
            raise credentials_exception
            
        if not user.is_active:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Inactive user"
            )
        
        return user

    async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
        """Get current active user"""
        if not current_user.is_active:
            raise HTTPException(status_code=400, detail="Inactive user")
        return current_user

Creating API Endpoints

Endpoints are the public interface of our API. FastAPI makes defining endpoints intuitive and powerful:

Decorators: Define HTTP method and route
Type hints: Automate validation and documentation
Dependency injection: Provides resources like DB and authenticated users
Response models: Control exactly what data is returned

Authentication Endpoints

Authentication endpoints handle the complete user management lifecycle. OAuth2PasswordRequestForm is the OAuth2 standard for username/password login.


    # app/api/v1/auth.py
    from datetime import timedelta
    from fastapi import APIRouter, Depends, HTTPException, status
    from fastapi.security import OAuth2PasswordRequestForm
    from sqlalchemy.orm import Session

    from app.api.deps import get_db
    from app.core.config import settings
    from app.core.security import create_access_token
    from app.schemas.user import UserCreate, User, Token
    from app.services.auth_service import AuthService

    router = APIRouter()

    @router.post("/register", response_model=User, status_code=status.HTTP_201_CREATED)
    def register_user(user: UserCreate, db: Session = Depends(get_db)):
        """Register new user"""
        
        # Check if username already exists
        if AuthService.is_username_taken(db, user.username):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Username already registered"
            )
        
        # Check if email already exists
        if AuthService.is_email_taken(db, user.email):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Email already registered"
            )
        
        # Create user
        return AuthService.create_user(db=db, user=user)

    @router.post("/login", response_model=Token)
    def login_user(
        form_data: OAuth2PasswordRequestForm = Depends(),
        db: Session = Depends(get_db)
    ):
        """User login"""
        
        # Authenticate user
        user = AuthService.authenticate_user(db, form_data.username, form_data.password)
        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Incorrect username or password",
                headers={"WWW-Authenticate": "Bearer"},
            )
        
        # Create access token
        access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"sub": user.username}, expires_delta=access_token_expires
        )
        
        return {"access_token": access_token, "token_type": "bearer"}

    @router.get("/me", response_model=User)
    def read_users_me(current_user: User = Depends(get_current_active_user)):
        """Get current user information"""
        return current_user

Task Endpoints

This set of endpoints implements a complete CRUD with advanced features:

Pagination: To handle large volumes of data
Filtering: By completion status and priority
Data isolation: Each user only sees their tasks
Atomic operations: Like toggle to change state

Parameter validation with Query() provides automatic documentation and type validation.


    # app/api/v1/tasks.py
    from typing import List
    from fastapi import APIRouter, Depends, HTTPException, status, Query
    from sqlalchemy.orm import Session

    from app.api.deps import get_db, get_current_active_user
    from app.models.user import User
    from app.schemas.task import Task, TaskCreate, TaskUpdate
    from app.services.task_service import TaskService

    router = APIRouter()

    @router.post("/", response_model=Task, status_code=status.HTTP_201_CREATED)
    def create_task(
        task: TaskCreate,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Create new task"""
        return TaskService.create_task(db=db, task=task, user_id=current_user.id)

    @router.get("/", response_model=List[Task])
    def read_tasks(
        skip: int = Query(0, ge=0, description="Number of tasks to skip"),
        limit: int = Query(100, ge=1, le=100, description="Maximum number of tasks to return"),
        completed: bool = Query(None, description="Filter by completion status"),
        priority: str = Query(None, description="Filter by priority (low, medium, high)"),
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Get current user's tasks with optional filters"""
        return TaskService.get_user_tasks(
            db=db, 
            user_id=current_user.id,
            skip=skip,
            limit=limit,
            completed=completed,
            priority=priority
        )

    @router.get("/{task_id}", response_model=Task)
    def read_task(
        task_id: int,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Get specific task"""
        task = TaskService.get_task(db=db, task_id=task_id, user_id=current_user.id)
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task

    @router.put("/{task_id}", response_model=Task)
    def update_task(
        task_id: int,
        task_update: TaskUpdate,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Update task"""
        task = TaskService.update_task(
            db=db, 
            task_id=task_id, 
            task_update=task_update, 
            user_id=current_user.id
        )
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task

    @router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
    def delete_task(
        task_id: int,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Delete task"""
        success = TaskService.delete_task(db=db, task_id=task_id, user_id=current_user.id)
        if not success:
            raise HTTPException(status_code=404, detail="Task not found")

    @router.patch("/{task_id}/toggle", response_model=Task)
    def toggle_task_completion(
        task_id: int,
        current_user: User = Depends(get_current_active_user),
        db: Session = Depends(get_db)
    ):
        """Toggle task completion status"""
        task = TaskService.toggle_task_completion(
            db=db, 
            task_id=task_id, 
            user_id=current_user.id
        )
        if task is None:
            raise HTTPException(status_code=404, detail="Task not found")
        return task

Task Service

The TaskService implements all business logic for task management:

Encapsulation: All DB logic in one place
Reusability: Methods that can be used from different endpoints
Transactions: Safe handling of DB operations
Dynamic filtering: Query building based on parameters

Using exclude_unset=True in updates allows partial modifications.


    # app/services/task_service.py
    from typing import List, Optional
    from sqlalchemy.orm import Session
    from sqlalchemy import and_

    from app.models.task import Task
    from app.schemas.task import TaskCreate, TaskUpdate

    class TaskService:
        
        @staticmethod
        def create_task(db: Session, task: TaskCreate, user_id: int) -> Task:
            """Create new task"""
            db_task = Task(**task.dict(), owner_id=user_id)
            db.add(db_task)
            db.commit()
            db.refresh(db_task)
            return db_task
        
        @staticmethod
        def get_task(db: Session, task_id: int, user_id: int) -> Optional[Task]:
            """Get specific user task"""
            return db.query(Task).filter(
                and_(Task.id == task_id, Task.owner_id == user_id)
            ).first()
        
        @staticmethod
        def get_user_tasks(
            db: Session, 
            user_id: int, 
            skip: int = 0, 
            limit: int = 100,
            completed: Optional[bool] = None,
            priority: Optional[str] = None
        ) -> List[Task]:
            """Get user tasks with filters"""
            query = db.query(Task).filter(Task.owner_id == user_id)
            
            # Apply optional filters
            if completed is not None:
                query = query.filter(Task.is_completed == completed)
            
            if priority is not None:
                query = query.filter(Task.priority == priority)
            
            return query.offset(skip).limit(limit).all()
        
        @staticmethod
        def update_task(
            db: Session, 
            task_id: int, 
            task_update: TaskUpdate, 
            user_id: int
        ) -> Optional[Task]:
            """Update task"""
            db_task = TaskService.get_task(db, task_id, user_id)
            if db_task is None:
                return None
            
            # Update only provided fields
            update_data = task_update.dict(exclude_unset=True)
            for field, value in update_data.items():
                setattr(db_task, field, value)
            
            db.commit()
            db.refresh(db_task)
            return db_task
        
        @staticmethod
        def delete_task(db: Session, task_id: int, user_id: int) -> bool:
            """Delete task"""
            db_task = TaskService.get_task(db, task_id, user_id)
            if db_task is None:
                return False
            
            db.delete(db_task)
            db.commit()
            return True
        
        @staticmethod
        def toggle_task_completion(db: Session, task_id: int, user_id: int) -> Optional[Task]:
            """Toggle completion status"""
            db_task = TaskService.get_task(db, task_id, user_id)
            if db_task is None:
                return None
            
            db_task.is_completed = not db_task.is_completed
            db.commit()
            db.refresh(db_task)
            return db_task

Main Application and Configuration

Main Application Configuration

The main.py file is our application’s entry point. Here we configure:

Middleware: For CORS, security, logging, etc.
Routers: Modular organization of endpoints
Documentation: OpenAPI/Swagger configuration
Initialization: Table creation and initial setup

CORS is especially important to allow frontends on different domains to consume our API.


    # app/main.py
    from fastapi import FastAPI
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.middleware.trustedhost import TrustedHostMiddleware

    from app.core.config import settings
    from app.api.v1 import auth, tasks
    from app.db.database import engine
    from app.models import user, task

    # Create tables in database
    user.Base.metadata.create_all(bind=engine)
    task.Base.metadata.create_all(bind=engine)

    # Create FastAPI instance
    app = FastAPI(
        title=settings.PROJECT_NAME,
        version=settings.PROJECT_VERSION,
        description="A modern API for task management built with FastAPI and PostgreSQL",
        openapi_url=f"{settings.API_V1_STR}/openapi.json"
    )

    # Configure CORS
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.BACKEND_CORS_ORIGINS,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Additional security middleware
    app.add_middleware(TrustedHostMiddleware, allowed_hosts=["*"])

    # Include routers
    app.include_router(auth.router, prefix=f"{settings.API_V1_STR}/auth", tags=["authentication"])
    app.include_router(tasks.router, prefix=f"{settings.API_V1_STR}/tasks", tags=["tasks"])

    # Health endpoint
    @app.get("/", tags=["health"])
    def health_check():
        """Health check endpoint"""
        return {
            "message": "Task Manager API is running",
            "version": settings.PROJECT_VERSION,
            "status": "healthy"
        }

    @app.get("/health", tags=["health"])
    def health():
        """Detailed health endpoint"""
        return {
            "status": "healthy",
            "service": settings.PROJECT_NAME,
            "version": settings.PROJECT_VERSION
        }

Environment Variables File

Create .env.example with the necessary configuration:


    # .env.example
    # Application configuration
    PROJECT_NAME="Task Manager API"
    PROJECT_VERSION="1.0.0"

    # Security
    SECRET_KEY="your-super-secret-key-change-this-in-production"
    ACCESS_TOKEN_EXPIRE_MINUTES=30

    # Database
    DATABASE_URL="postgresql://postgres:postgres@localhost:5432/taskmanager"

    # CORS
    BACKEND_CORS_ORIGINS=["http://localhost:3000","http://localhost:8080"]

Automated Testing

Testing is crucial for maintaining code quality. FastAPI has excellent support for testing with pytest:

TestClient: Simulates HTTP requests without starting a server
Fixtures: Reusable configuration for tests
Dependency overrides: Allows using in-memory DB for tests
Isolation: Each test has a clean DB

Test Configuration

We use SQLite in memory for tests because it’s fast and resets automatically. Dependency override allows replacing the real DB with the testing one.


    # tests/conftest.py
    import pytest
    from fastapi.testclient import TestClient
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from app.main import app
    from app.db.database import get_db, Base
    from app.core.config import settings

    # In-memory database for tests
    SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

    engine = create_engine(
        SQLALCHEMY_DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    def override_get_db():
        try:
            db = TestingSessionLocal()
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_get_db

    @pytest.fixture(scope="function")
    def db():
        Base.metadata.create_all(bind=engine)
        yield
        Base.metadata.drop_all(bind=engine)

    @pytest.fixture(scope="function")
    def client(db):
        with TestClient(app) as test_client:
            yield test_client

    @pytest.fixture
    def user_data():
        return {
            "username": "testuser",
            "email": "test@example.com",
            "password": "testpassword123"
        }

    @pytest.fixture
    def authenticated_client(client, user_data):
        # Register user
        client.post("/api/v1/auth/register", json=user_data)
        
        # Login
        response = client.post(
            "/api/v1/auth/login",
            data={"username": user_data["username"], "password": user_data["password"]}
        )
        token = response.json()["access_token"]
        
        # Configure authentication headers
        client.headers.update({"Authorization": f"Bearer {token}"})
        return client

Authentication Tests

These tests verify the complete authentication flow:

User registration with duplicate validation
Login with correct and incorrect credentials
Access to protected endpoints with valid tokens

Each test is independent and verifies a specific aspect of the system.


    # tests/test_auth.py
    import pytest
    from fastapi.testclient import TestClient

    def test_register_user(client: TestClient, user_data):
        """Test user registration"""
        response = client.post("/api/v1/auth/register", json=user_data)
        
        assert response.status_code == 201
        data = response.json()
        assert data["username"] == user_data["username"]
        assert data["email"] == user_data["email"]
        assert "id" in data
        assert "hashed_password" not in data

    def test_register_duplicate_username(client: TestClient, user_data):
        """Test registration with duplicate username"""
        # First registration
        client.post("/api/v1/auth/register", json=user_data)
        
        # Second registration with same username
        response = client.post("/api/v1/auth/register", json=user_data)
        
        assert response.status_code == 400
        assert "Username already registered" in response.json()["detail"]

    def test_login_user(client: TestClient, user_data):
        """Test user login"""
        # Register user
        client.post("/api/v1/auth/register", json=user_data)
        
        # Login
        response = client.post(
            "/api/v1/auth/login",
            data={"username": user_data["username"], "password": user_data["password"]}
        )
        
        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert data["token_type"] == "bearer"

    def test_login_invalid_credentials(client: TestClient, user_data):
        """Test login with invalid credentials"""
        response = client.post(
            "/api/v1/auth/login",
            data={"username": "nonexistent", "password": "wrongpassword"}
        )
        
        assert response.status_code == 401
        assert "Incorrect username or password" in response.json()["detail"]

    def test_get_current_user(authenticated_client: TestClient, user_data):
        """Test getting current user information"""
        response = authenticated_client.get("/api/v1/auth/me")
        
        assert response.status_code == 200
        data = response.json()
        assert data["username"] == user_data["username"]
        assert data["email"] == user_data["email"]

Task Tests

Task tests verify the complete CRUD:

Creation with valid data
Listing with filters
Partial updates
Deletion and verification
Special operations like toggle

These tests ensure that business logic works correctly.


    # tests/test_tasks.py
    import pytest
    from fastapi.testclient import TestClient

    def test_create_task(authenticated_client: TestClient):
        """Test creating task"""
        task_data = {
            "title": "Test Task",
            "description": "This is a test task",
            "priority": "high"
        }
        
        response = authenticated_client.post("/api/v1/tasks/", json=task_data)
        
        assert response.status_code == 201
        data = response.json()
        assert data["title"] == task_data["title"]
        assert data["description"] == task_data["description"]
        assert data["priority"] == task_data["priority"]
        assert data["is_completed"] is False
        assert "id" in data

    def test_get_tasks(authenticated_client: TestClient):
        """Test getting task list"""
        # Create some tasks
        for i in range(3):
            authenticated_client.post("/api/v1/tasks/", json={
                "title": f"Task {i}",
                "description": f"Description {i}"
            })
        
        response = authenticated_client.get("/api/v1/tasks/")
        
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 3

    def test_get_task_by_id(authenticated_client: TestClient):
        """Test getting specific task"""
        # Create task
        task_response = authenticated_client.post("/api/v1/tasks/", json={
            "title": "Specific Task",
            "description": "Specific description"
        })
        task_id = task_response.json()["id"]
        
        # Get task
        response = authenticated_client.get(f"/api/v1/tasks/{task_id}")
        
        assert response.status_code == 200
        data = response.json()
        assert data["title"] == "Specific Task"

    def test_update_task(authenticated_client: TestClient):
        """Test updating task"""
        # Create task
        task_response = authenticated_client.post("/api/v1/tasks/", json={
            "title": "Original Title",
            "description": "Original description"
        })
        task_id = task_response.json()["id"]
        
        # Update task
        update_data = {"title": "Updated Title", "is_completed": True}
        response = authenticated_client.put(f"/api/v1/tasks/{task_id}", json=update_data)
        
        assert response.status_code == 200
        data = response.json()
        assert data["title"] == "Updated Title"
        assert data["is_completed"] is True

    def test_delete_task(authenticated_client: TestClient):
        """Test deleting task"""
        # Create task
        task_response = authenticated_client.post("/api/v1/tasks/", json={
            "title": "Task to Delete"
        })
        task_id = task_response.json()["id"]
        
        # Delete task
        response = authenticated_client.delete(f"/api/v1/tasks/{task_id}")
        
        assert response.status_code == 204
        
        # Verify it no longer exists
        get_response = authenticated_client.get(f"/api/v1/tasks/{task_id}")
        assert get_response.status_code == 404

    def test_toggle_task_completion(authenticated_client: TestClient):
        """Test toggling completion status"""
        # Create task
        task_response = authenticated_client.post("/api/v1/tasks/", json={
            "title": "Task to Toggle"
        })
        task_id = task_response.json()["id"]
        
        # Toggle to completed
        response = authenticated_client.patch(f"/api/v1/tasks/{task_id}/toggle")
        
        assert response.status_code == 200
        data = response.json()
        assert data["is_completed"] is True
        
        # Toggle back to not completed
        response = authenticated_client.patch(f"/api/v1/tasks/{task_id}/toggle")
        data = response.json()
        assert data["is_completed"] is False

Running Tests

To run our API tests, we need to install testing dependencies and run pytest:


    # Install testing dependencies (already included in requirements.txt)
    pip install pytest pytest-asyncio httpx

    # Run all tests
    pytest

    # Run tests with detailed output
    pytest -v

    # Run tests from a specific file
    pytest tests/test_auth.py

    # Run a specific test
    pytest tests/test_auth.py::test_register_user

    # Run tests with coverage
    pip install pytest-cov
    pytest --cov=app --cov-report=html

Tests run quickly because they use an in-memory SQLite database. Each test is independent and the database resets between tests.

Recommended Test Structure

For larger projects, organize tests as follows:

tests/
├── conftest.py              # Shared configuration
├── test_auth.py            # Authentication tests
├── test_tasks.py           # Task tests
├── test_database.py        # Model tests
└── integration/            # Integration tests
    └── test_full_flow.py   # End-to-end tests

Production Considerations

⚠️ Warning: This tutorial is designed for development and learning. To deploy to production, you need to implement additional security, performance, and monitoring considerations.

Critical Aspects for Production

Security:
- Change SECRET_KEY to a cryptographically generated one
- Use HTTPS with valid SSL certificates
- Implement rate limiting per IP and per user
- Validate and sanitize all inputs
- Configure restrictive CORS for specific domains
Database:
- Use a dedicated PostgreSQL server
- Implement automatic backups
- Configure connection pooling
- Optimize queries with appropriate indexes
Monitoring and logging:
- Implement structured logging
- Configure application metrics
- Alerts for errors and performance
- Health checks for containers
Scalability:
- Use multiple Uvicorn workers
- Implement load balancing
- Cache with Redis for frequent data
- CDN for static content
Configuration management:
- Use Kubernetes or Docker Swarm secrets
- Environment-specific variables
- Centralized configuration management

Testing the Complete API

Starting the Environment


    # Clone environment variables
    cp .env.example .env

    # Edit variables according to your environment
    nano .env

    # Start with Docker Compose
    docker-compose up -d

    # Verify it's working
    curl http://localhost:8000/health

Usage Examples with curl


    # 1. Register user
    curl -X POST "http://localhost:8000/api/v1/auth/register" \
        -H "Content-Type: application/json" \
        -d '{
        "username": "demo_user",
        "email": "demo@example.com",
        "password": "password123"
        }'

    # 2. Login and get token
    TOKEN=$(curl -X POST "http://localhost:8000/api/v1/auth/login" \
        -H "Content-Type: application/x-www-form-urlencoded" \
        -d "username=demo_user&password=password123" | jq -r '.access_token')

    # 3. Create task
    curl -X POST "http://localhost:8000/api/v1/tasks/" \
        -H "Authorization: Bearer $TOKEN" \
        -H "Content-Type: application/json" \
        -d '{
        "title": "Implement new feature",
        "description": "Develop reports module",
        "priority": "high",
        "due_date": "2025-10-01T10:00:00"
        }'

    # 4. List tasks
    curl -X GET "http://localhost:8000/api/v1/tasks/" \
        -H "Authorization: Bearer $TOKEN"

    # 5. Update task
    curl -X PUT "http://localhost:8000/api/v1/tasks/1" \
        -H "Authorization: Bearer $TOKEN" \
        -H "Content-Type: application/json" \
        -d '{
        "title": "Feature completed",
        "is_completed": true
        }'

Automatic Documentation

Once the application is running, you can access:

Swagger UI: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc
OpenAPI JSON: http://localhost:8000/api/v1/openapi.json

FastAPI automatically generates this documentation based on the type hints and docstrings of your endpoints. The interface is completely interactive and allows testing the API directly from the browser.

Conclusions and Next Steps

You’ve built a complete and modern REST API following current best practices. The project includes:

✅ Secure JWT authentication with user registration and login
✅ Complete CRUD for task management with advanced filters
✅ Automatic data validation with Pydantic
✅ Interactive documentation generated automatically
✅ Automated testing with high coverage
✅ Containerization with Docker for easy deployment
✅ Production configuration with Nginx and SSL

Advantages of the Chosen Stack

FastAPI has proven to be an excellent choice for several reasons:

Developer Experience: Autocompletion, real-time error detection
Automatic documentation: Swagger UI and ReDoc with no additional effort
Performance: Comparable to Go and Node.js frameworks
Modernity: Takes advantage of Python’s latest features

Integration with PostgreSQL provides:

Robustness: Complete ACID database with excellent performance
Scalability: Handles from startups to enterprise
Ecosystem: Mature extensions and tools

Docker ensures:

Consistency: Same environment in development, testing, and production
Simplicity: One command to start the entire stack
Isolation: Encapsulated dependencies without conflicts

Static typing with Pydantic:

Prevents errors: Automatic validation in development
Improves maintainability: Safe refactoring with IDEs
Documents code: Types are living documentation

Automated tests:

Confidence: Refactoring without fear of breaking functionality
Documentation: Tests show how to use the API
Quality: Detect regressions before reaching production

Possible Improvements and Extensions

Caching with Redis to improve performance
Structured logging with tools like ELK Stack
Advanced rate limiting per user
Push and email notifications
More sophisticated filtering and search
API versioning to maintain compatibility
Metrics and monitoring with Prometheus
Automatic database backup

Additional Resources

FastAPI Documentation: https://fastapi.tiangolo.com
PostgreSQL Best Practices: PostgreSQL Performance Tips
Docker Deployment Guide: Docker in Production

This API can serve as a solid foundation for more complex projects, from MVPs to enterprise applications. The architectural pattern and technical decisions are scalable and aligned with industry best practices in 2025.

Did you like the tutorial? Share your experience implementing the API or tell us what functionalities you would add for your specific use case.

Alberto Méndez

Master REST APIs with FastAPI and PostgreSQL: End-to-end tutorial with source code

Sep 15, 2025

Why FastAPI for Modern APIs?

Exceptional Performance

Automatic Documentation

Modern Development with Type Hints

Native Asynchronous Compatibility

Setting Up the Development Environment

Project Structure

Project Dependencies

Docker Configuration

Configuration and Data Models

Configuration System

Database Connection

Data Models

Pydantic Schemas

Implementing JWT Authentication

Security System

Authentication Service

Authentication Dependencies

Creating API Endpoints

Authentication Endpoints

Task Endpoints

Task Service

Main Application and Configuration

Main Application Configuration

Environment Variables File

Automated Testing

Test Configuration

Authentication Tests

Task Tests

Running Tests

Recommended Test Structure

Production Considerations

Critical Aspects for Production

Testing the Complete API

Starting the Environment

Usage Examples with curl

Automatic Documentation

Conclusions and Next Steps

Advantages of the Chosen Stack

Possible Improvements and Extensions

Additional Resources

Alberto Méndez

Related posts

Master Docker containers, volumes and networks

Creating a REST API with Express and Docker

Databases, an introduction to fundamentals

Our work

Google Analytics

Lightbox

Disqus

Youtube

Cookies