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_enginefrom sqlalchemy.orm import sessionmaker, declarative_baseDATABASE_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.pyfrom sqlalchemy import Column, Integer, Stringfrom app.database.connection import Baseclass User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True)username = Column(String, unique=True)password = Column(String)
models/task.pyfrom sqlalchemy import Column, Integer, String, ForeignKeyfrom app.database.connection import Baseclass 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.pyfrom pydantic import BaseModelclass UserCreate(BaseModel):username: strpassword: str
schemas/task.pyclass TaskCreate(BaseModel):title: strStep 4: Authentication (JWT)
utils/auth.pyfrom jose import jwtfrom datetime import datetime, timedeltaSECRET_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.pyfrom fastapi import APIRouter, Dependsfrom sqlalchemy.orm import Sessionfrom app.database.connection import SessionLocalfrom app.models.user import Userfrom app.schemas.user import UserCreatefrom app.utils.auth import create_tokenrouter = 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.pyfrom fastapi import APIRouter, Dependsfrom sqlalchemy.orm import Sessionfrom app.database.connection import SessionLocalfrom app.models.task import Taskfrom app.schemas.task import TaskCreaterouter = 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 FastAPIfrom app.routes import auth, taskfrom app.database.connection import Base, engineapp = 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 hashing2. Add token verification3. Use user-specific tasks4. Add validation & error handlingBest Practices
1. Use layered architecture2. Separate routes, models, schemas3. Secure authentication4. Write clean reusable code
Comments
Post a Comment