FastAPI Blog: Build A REST API For Your Blog
Hey guys! Are you thinking about creating your own blog and want to build it with modern tools? Let's dive into how to build a blog using FastAPI, a fantastic and high-performance Python web framework for building APIs. This guide will walk you through setting up your project, designing your database models, creating API endpoints, and testing your application.
Setting Up Your FastAPI Project
First things first, let's get our development environment ready. Make sure you have Python installed. I would recommend using Python 3.8 or higher. Then, we'll create a virtual environment to keep our project dependencies isolated. Open your terminal and run these commands:
python3 -m venv venv
source venv/bin/activate # On Windows, use `venv\Scripts\activate`
pip install fastapi uvicorn sqlalchemy python-multipart alembic
FastAPI itself is the web framework, Uvicorn is an ASGI server to run our application, SQLAlchemy is an ORM to interact with our database, python-multipart is needed for handling file uploads, and Alembic helps us manage database migrations. These are key ingredients, trust me!
Now, let's create our main application file, main.py:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"message": "Hello World"}
This is the most basic FastAPI application. To run it, use the following command:
uvicorn main:app --reload
Open your browser and go to http://127.0.0.1:8000. You should see the "Hello World" message. The --reload flag means the server will automatically restart whenever you make changes to your code. Super handy, right?
Designing Your Database Models
A blog needs a database to store posts, users, and other data. We’ll use SQLAlchemy to define our database models. Let’s start by setting up the database connection. Create a file named database.py:
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./blog.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Here, we are using SQLite for simplicity. In a production environment, you might want to use PostgreSQL or MySQL. The SQLALCHEMY_DATABASE_URL variable defines the connection string. We create a database engine and a session factory, which we'll use to interact with the database. Base is used for defining our models.
Now, let’s define our blog post model. Create a file named models.py:
from sqlalchemy import Boolean, Column, Integer, String, DateTime
from sqlalchemy.sql import func
from database import Base
class BlogPost(Base):
__tablename__ = "blog_posts"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
content = Column(String)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
is_published = Column(Boolean, default=False)
This model defines a BlogPost table with columns for id, title, content, created_at, updated_at, and is_published. Each post will have a unique ID, a title, the main content, timestamps for creation and updates, and a boolean field to indicate if the post is published. Don't forget to import Base from database.py.
Next, we need to create the database tables. We'll use Alembic for this. First, initialize Alembic:
alembic init alembic
This creates an alembic directory with configuration files. Now, edit the alembic.ini file and set the sqlalchemy.url to your database URL:
sqlalchemy.url = sqlite:///./blog.db
Then, edit the env.py file in the alembic directory. Add the following lines to import your models:
import sys
sys.path.append('.')
from database import Base
from models import BlogPost # Import your models here
target_metadata = Base.metadata
Now, generate a migration:
alembic revision --autogenerate -m "Create blog_posts table"
This creates a new migration script based on the changes in your models. Finally, apply the migration:
alembic upgrade head
This will create the blog_posts table in your database. Now we have our database set up and our model defined, making us ready to perform CRUD operations.
Creating API Endpoints
Now, let's create the API endpoints to perform CRUD (Create, Read, Update, Delete) operations on our blog posts. We’ll start by defining the data models using Pydantic. Create a file named schemas.py:
from datetime import datetime
from pydantic import BaseModel
class BlogPostBase(BaseModel):
title: str
content: str
class BlogPostCreate(BlogPostBase):
pass
class BlogPostUpdate(BlogPostBase):
is_published: bool
class BlogPost(BlogPostBase):
id: int
created_at: datetime
updated_at: datetime | None
is_published: bool
class Config:
orm_mode = True
Here, BlogPostBase defines the basic attributes of a blog post. BlogPostCreate is used for creating new posts, BlogPostUpdate for updating existing ones, and BlogPost includes the id and timestamps. orm_mode = True is important because it tells Pydantic to read data from SQLAlchemy models.
Now, let's add the API endpoints to main.py:
from typing import List
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
import models
import schemas
from database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/posts/", response_model=schemas.BlogPost)
async def create_post(post: schemas.BlogPostCreate, db: Session = Depends(get_db)):
db_post = models.BlogPost(**post.dict())
db.add(db_post)
db.commit()
db.refresh(db_post)
return db_post
@app.get("/posts/", response_model=List[schemas.BlogPost])
async def read_posts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
posts = db.query(models.BlogPost).offset(skip).limit(limit).all()
return posts
@app.get("/posts/{post_id}", response_model=schemas.BlogPost)
async def read_post(post_id: int, db: Session = Depends(get_db)):
post = db.query(models.BlogPost).filter(models.BlogPost.id == post_id).first()
if post is None:
raise HTTPException(status_code=404, detail="Post not found")
return post
@app.put("/posts/{post_id}", response_model=schemas.BlogPost)
async def update_post(post_id: int, post: schemas.BlogPostUpdate, db: Session = Depends(get_db)):
db_post = db.query(models.BlogPost).filter(models.BlogPost.id == post_id).first()
if db_post is None:
raise HTTPException(status_code=404, detail="Post not found")
for key, value in post.dict(exclude_unset=True).items():
setattr(db_post, key, value)
db.add(db_post)
db.commit()
db.refresh(db_post)
return db_post
@app.delete("/posts/{post_id}", response_model=schemas.BlogPost)
async def delete_post(post_id: int, db: Session = Depends(get_db)):
post = db.query(models.BlogPost).filter(models.BlogPost.id == post_id).first()
if post is None:
raise HTTPException(status_code=404, detail="Post not found")
db.delete(post)
db.commit()
return post
We define a get_db function as a dependency to manage database sessions. The /posts/ endpoint with a POST method creates a new post. The /posts/ endpoint with a GET method retrieves a list of posts. The /posts/{post_id} endpoint with a GET method retrieves a specific post by ID. The /posts/{post_id} endpoint with a PUT method updates an existing post. And finally, the /posts/{post_id} endpoint with a DELETE method deletes a post. Each endpoint uses the schemas to validate the data and returns a BlogPost object.
Testing Your Application
Testing is crucial. FastAPI has excellent support for testing using pytest. First, install pytest and httpx:
pip install pytest httpx
Create a file named test_main.py:
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import pytest
from main import app, get_db
from database import Base
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture()
def test_db():
Base.metadata.create_all(bind=engine)
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
Base.metadata.drop_all(bind=engine)
@pytest.fixture()
def client(test_db):
def override_get_db():
try:
yield test_db
finally:
test_db.close()
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
yield client
app.dependency_overrides = {}
def test_create_post(client):
response = client.post(
"/posts/",
json={"title": "Test Post", "content": "This is a test post."},
)
assert response.status_code == 200
data = response.json()
assert data["title"] == "Test Post"
assert data["content"] == "This is a test post."
assert "id" in data
def test_read_posts(client):
response = client.get("/posts/")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
This test suite sets up a separate testing database and overrides the get_db dependency to use the test database. test_create_post tests the creation of a new post, and test_read_posts tests the retrieval of all posts. Run the tests using:
pytest
Make sure all tests pass. If not, fix them before moving on. These are simple tests, but they give you a starting point for writing more comprehensive tests.
Conclusion
Alright, guys! You've just built a basic blog API using FastAPI. You've set up your project, designed your database models with SQLAlchemy, created API endpoints for CRUD operations, and written tests. This is just the beginning. You can extend this application by adding user authentication, comments, categories, and more. FastAPI makes it easy to build robust and scalable APIs. Happy coding!
