Build a Complete FastAPI Project (CRUD + Auth + Database)

 

Introduction

So far, you’ve learned all the core concepts of FastAPI.Now it’s time to combine everything into a real-world project. In this blog, we’ll build a production-style FastAPI application with:

  • CRUD operations
  • JWT Authentication
  • Database (SQLAlchemy)
  • Clean project structure

Project Overview

We’ll build a simple Task Management API where users can:

  • Register & login
  • Create tasks
  • View tasks
  • Update & delete tasks

Project Structure

fastapi-project/
├── app/
│ ├── main.py
│ ├── database/
│ ├── models/
│ ├── schemas/
│ ├── routes/
│ ├── dependencies/
│ └── utils/

Step 1: Database Setup

database/connection.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

Step 2: Models

models/user.py

from sqlalchemy import Column, Integer, String
from app.database.connection import Base

class User(Base):
__tablename__ = "users"

id = Column(Integer, primary_key=True)
username = Column(String, unique=True)
password = Column(String)


models/task.py

from sqlalchemy import Column, Integer, String, ForeignKey
from app.database.connection import Base

class Task(Base):
__tablename__ = "tasks"

id = Column(Integer, primary_key=True)
title = Column(String)
user_id = Column(Integer, ForeignKey("users.id"))

Step 3: Schemas

schemas/user.py

from pydantic import BaseModel

class UserCreate(BaseModel):
username: str
password: str

schemas/task.py

class TaskCreate(BaseModel):
title: str


Step 4: Authentication (JWT)

utils/auth.py

from jose import jwt
from datetime import datetime, timedelta

SECRET_KEY = "secret"
ALGORITHM = "HS256"

def create_token(data: dict):
data.update({"exp": datetime.utcnow() + timedelta(minutes=60)})
return jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM)


Step 5: Login & Register

routes/auth.py

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.database.connection import SessionLocal
from app.models.user import User
from app.schemas.user import UserCreate
from app.utils.auth import create_token

router = APIRouter() def get_db(): db = SessionLocal() try: yield db finally: db.close() @router.post("/register") def register(user: UserCreate, db: Session = Depends(get_db)): new_user = User(username=user.username, password=user.password) db.add(new_user) db.commit() return {"message": "User created"} @router.post("/login") def login(user: UserCreate, db: Session = Depends(get_db)): db_user = db.query(User).filter(User.username == user.username).first() if not db_user: return {"error": "Invalid user"} token = create_token({"sub": user.username})
return {"access_token": token}


Step 6: Task CRUD (Protected)

routes/task.py

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.database.connection import SessionLocal
from app.models.task import Task
from app.schemas.task import TaskCreate

router = APIRouter() def get_db(): db = SessionLocal() try: yield db finally: db.close() @router.post("/tasks/") def create_task(task: TaskCreate, db: Session = Depends(get_db)): db_task = Task(title=task.title, user_id=1) db.add(db_task) db.commit() return db_task @router.get("/tasks/") def get_tasks(db: Session = Depends(get_db)):
return db.query(Task).all()


Step 7: Main App

main.py

from fastapi import FastAPI
from app.routes import auth, task
from app.database.connection import Base, engine

app = FastAPI()

Base.metadata.create_all(bind=engine)

app.include_router(auth.router)
app.include_router(task.router)


Test Your API

Endpoints:

  • POST /register
  • POST /login
  • POST /tasks/
  • GET /tasks/

Improvements (Production Level)

1. Add password hashing
2. Add token verification
3. Use user-specific tasks
4. Add validation & error handling

Best Practices

1. Use layered architecture
2. Separate routes, models, schemas
3. Secure authentication
4. Write clean reusable code


Comments

Popular posts from this blog

Database Integration in FastAPI (SQLAlchemy CRUD)

Middleware & CORS in FastAPI

Python Data Handling