Quick Review and Final Project Structure
1.1 Quick Review: Where We Are
Welcome to the final lesson of our Python and FastAPI journey! Over the past 19 lessons, you have built a solid foundation in Python and learned how to create a REST API using FastAPI. You have mastered variables, loops, functions, object-oriented programming, and even connected your API to a SQLite database. Now, in this final chapter, we will bring everything together to build a complete, production-ready Tasks API project.
Before we start coding, let's do a quick review of the key concepts we have covered:
- Python Basics: Variables, data types, conditionals, loops, functions, lists, dictionaries, file handling, and error handling.
- Object-Oriented Programming (OOP): Classes, objects, and methods.
- FastAPI Fundamentals: Creating endpoints, path parameters, query parameters, Pydantic models, and CRUD operations.
- Database Integration: Using SQLite with SQLAlchemy to store and retrieve tasks.
- Testing: Using Swagger UI and curl to test your API.
In this lesson, we will not learn new concepts. Instead, we will apply everything you have learned to build a final, structured project that you can extend in the future.
1.2 Why a Final Project Structure Matters
When you build a real-world API, you cannot just throw all your code into one file. A good project structure makes your code:
- Organized: Easy to find and modify specific parts.
- Scalable: You can add new features without breaking existing ones.
- Maintainable: Other developers (or future you) can understand and update the code.
In this final project, we will use a clean folder structure that separates concerns: models, database logic, API routes, and the main application file.
1.3 Final Project Structure
Here is the folder structure we will use for our Tasks API:
tasks_api/
│
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
├── requirements.txt
└── __init__.py
Let's explain each file:
- main.py: The entry point of your application. It creates the FastAPI app and includes the routers.
- database.py: Handles the database connection and session management using SQLAlchemy.
- models.py: Defines the SQLAlchemy models (database tables).
- schemas.py: Defines the Pydantic models for request and response validation.
- crud.py: Contains the CRUD (Create, Read, Update, Delete) functions that interact with the database.
- requirements.txt: Lists all the Python packages needed for the project.
1.4 Step-by-Step Implementation
Let's build each file one by one. We will start with the database setup.
Step 1: Create the database.py file
This file sets up the SQLite database connection and provides a session generator.
# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Explanation:
- We create an engine that connects to a local SQLite file named
tasks.db. SessionLocalis a factory for database sessions.Baseis the declarative base class for our models.get_db()is a dependency that provides a database session and ensures it is closed after use.
Step 2: Create the models.py file
This file defines the Task table in the database.
# models.py
from sqlalchemy import Column, Integer, String, Boolean
from database import Base
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, default="")
completed = Column(Boolean, default=False)
Explanation:
- We define a
Taskclass that inherits fromBase. - The table has four columns:
id,title,description, andcompleted. idis the primary key and will be auto-incremented.
Step 3: Create the schemas.py file
This file defines Pydantic models for data validation.
# schemas.py
from pydantic import BaseModel
class TaskBase(BaseModel):
title: str
description: str = ""
completed: bool = False
class TaskCreate(TaskBase):
pass
class Task(TaskBase):
id: int
class Config:
orm_mode = True
Explanation:
TaskBasecontains the common fields.TaskCreateis used when creating a new task (it inherits all fields from TaskBase).Taskincludes theidfield and setsorm_mode = Trueso that FastAPI can convert SQLAlchemy objects to Pydantic models automatically.
Step 4: Create the crud.py file
This file contains the functions that interact with the database.
# crud.py
from sqlalchemy.orm import Session
import models, schemas
def get_task(db: Session, task_id: int):
return db.query(models.Task).filter(models.Task.id == task_id).first()
def get_tasks(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.Task).offset(skip).limit(limit).all()
def create_task(db: Session, task: schemas.TaskCreate):
db_task = models.Task(**task.dict())
db.add(db_task)
db.commit()
db.refresh(db_task)
return db_task
def update_task(db: Session, task_id: int, task: schemas.TaskCreate):
db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
if db_task:
for key, value in task.dict().items():
setattr(db_task, key, value)
db.commit()
db.refresh(db_task)
return db_task
def delete_task(db: Session, task_id: int):
db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
if db_task:
db.delete(db_task)
db.commit()
return db_task
Explanation:
- Each function takes a database session and performs a specific operation.
create_taskcreates a new Task object from the Pydantic schema, adds it to the session, commits, and refreshes to get the generated ID.update_taskupdates only the fields that are provided.delete_taskremoves the task from the database.
Step 5: Create the main.py file
This is the main application file that ties everything together.
# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
import models, schemas, crud
from database import engine, get_db
models.Base.metadata.create_all(bind=engine)
app = FastAPI(title="Tasks API", description="A complete Tasks API built with FastAPI", version="1.0.0")
@app.post("/tasks/", response_model=schemas.Task)
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
return crud.create_task(db=db, task=task)
@app.get("/tasks/", response_model=list[schemas.Task])
def read_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
tasks = crud.get_tasks(db, skip=skip, limit=limit)
return tasks
@app.get("/tasks/{task_id}", response_model=schemas.Task)
def read_task(task_id: int, db: Session = Depends(get_db)):
db_task = crud.get_task(db, task_id=task_id)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
return db_task
@app.put("/tasks/{task_id}", response_model=schemas.Task)
def update_task(task_id: int, task: schemas.TaskCreate, db: Session = Depends(get_db)):
db_task = crud.update_task(db, task_id=task_id, task=task)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
return db_task
@app.delete("/tasks/{task_id}", response_model=schemas.Task)
def delete_task(task_id: int, db: Session = Depends(get_db)):
db_task = crud.delete_task(db, task_id=task_id)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
return db_task
Explanation:
- We create the database tables by calling
models.Base.metadata.create_all(bind=engine). - We define five endpoints: POST to create, GET to list all, GET by ID, PUT to update, and DELETE to remove.
- Each endpoint uses the
get_dbdependency to get a database session. - We use
response_modelto automatically validate and serialize the response. - If a task is not found, we raise an HTTP 404 exception.
Step 6: Create the requirements.txt file
List all the packages needed for this project.
fastapi
uvicorn
sqlalchemy
pydantic
1.5 Running the Final Project
To run your final Tasks API, follow these steps:
- Open a terminal in the
tasks_apifolder. - Install the required packages:
pip install -r requirements.txt - Run the server:
uvicorn main:app --reload - Open your browser and go to
http://127.0.0.1:8000/docsto see the interactive Swagger documentation.
You can now test all CRUD operations directly from the browser!
1.6 Common Mistakes and How to Avoid Them
- Forgetting to install dependencies: Always run
pip install -r requirements.txtbefore running the project. - Not activating the virtual environment: Make sure you are in the correct virtual environment to avoid package conflicts.
- Typo in model or schema fields: Double-check that field names match between models, schemas, and CRUD functions.
- Not handling the 404 case: Always check if a task exists before returning it, or you will get an error.
1.7 Practice Task
Now it's your turn to extend the project. Add a new endpoint that allows you to mark a task as completed without providing the full task data. The endpoint should be a PATCH request to /tasks/{task_id}/complete and should set the completed field to True.
Hint: You will need to create a new Pydantic schema for the partial update, add a new function in crud.py, and add a new route in main.py.
Once you finish, test it using Swagger UI. Congratulations on completing the course!
Loading ratings...