FastAPI Blog: Build A REST API With Security & Testing

S.Skip 127 views
FastAPI Blog: Build A REST API With Security & Testing

FastAPI Blog: Build a REST API with Security & Testing

Hey guys! Today, let’s dive into building a robust blog API using FastAPI. This isn’t just about slapping together some endpoints; we’re talking about creating a secure , testable , and maintainable application. FastAPI is perfect for this because it’s fast, easy to use, and comes with built-in support for things like data validation and automatic API documentation. So, grab your favorite beverage, fire up your code editor, and let’s get started!

Table of Contents

Setting Up Your FastAPI Project

First things first, you’ll need to set up a new project and install FastAPI along with its dependencies. I usually recommend using a virtual environment to keep your project dependencies isolated. This prevents conflicts with other Python projects you might be working on. To create a virtual environment, you can use venv : python3 -m venv venv . Activate it with source venv/bin/activate on Linux/macOS or venv\Scripts\activate on Windows. Once your virtual environment is active, you can install FastAPI and Uvicorn, an ASGI server that we’ll use to run our application. Use pip to install these packages: pip install fastapi uvicorn . These are the foundational elements. FastAPI provides the framework for building our API, and Uvicorn will serve our application.

Now that you have FastAPI installed, create a file named main.py . This will be the entry point of our application. Inside main.py , import FastAPI and create an instance of the FastAPI class: from fastapi import FastAPI. app = FastAPI(). This initializes our FastAPI application. Next, let’s define our first endpoint. We’ll start with a simple root endpoint that returns a welcome message. Use the @app.get("/") decorator to define a GET endpoint for the root path. Inside the function, return a dictionary with a message: “`python from fastapi import FastAPI

app = FastAPI()

@app.get(”/“) async def read_root(): return {“message”: “Welcome to the Awesome Blog API!”} 

This simple example demonstrates how easy it is to create endpoints with FastAPI. The `@app.get()` decorator specifies the HTTP method and path for the endpoint, and the function defines the logic to be executed when the endpoint is accessed. The `async` keyword indicates that this is an asynchronous function, which is a key feature of FastAPI that allows it to handle requests concurrently and efficiently. To run your application, use the command `uvicorn main:app --reload`. This starts the Uvicorn server, tells it to run the `app` instance from the `main.py` file, and enables the `--reload` flag, which automatically reloads the server whenever you make changes to your code. Open your browser and navigate to `http://127.0.0.1:8000` (or the address Uvicorn provides). You should see the JSON response: `{"message": "Welcome to the Awesome Blog API!"}`. Congratulations, you've just created your first FastAPI endpoint!

## Defining Blog Post Models with Pydantic

Next, we need to define the structure of our blog posts. We'll use Pydantic, which comes integrated with FastAPI, to create data models. Pydantic models are Python classes that define the structure and validation rules for our data. Create a new file named `models.py` and define a `BlogPost` model. This model will include fields like `id`, `title`, `content`, and `publication_date`. Here’s an example:

```python
from typing import Optional
from datetime import datetime
from pydantic import BaseModel

class BlogPost(BaseModel):
 id: Optional[int] = None
 title: str
 content: str
 publication_date: Optional[datetime] = None

In this model, BaseModel is imported from Pydantic, and it serves as the base class for our data model. id is defined as an optional integer (Optional[int] = None), meaning it can be None if the blog post hasn’t been assigned an ID yet (e.g., before it’s stored in a database). title and content are defined as strings (str), meaning they are required fields. publication_date is an optional datetime object, which can be None if the publication date is not yet set. Pydantic provides built-in data validation. If the incoming data does not match the defined types, Pydantic will raise a validation error. This helps ensure that your application receives and processes valid data. Now, let’s define another model for creating new blog posts. This model will be similar to BlogPost , but without the id field, as the ID will be automatically generated when the blog post is created. Here’s the CreateBlogPost model: 

from datetime import datetime
from pydantic import BaseModel

class CreateBlogPost(BaseModel):
 title: str
 content: str
 publication_date: datetime = datetime.now()

This model includes title and content as required strings, and publication_date as a datetime object with a default value of the current date and time (datetime.now()). This ensures that if the publication date is not explicitly provided, it will default to the time when the blog post is created. Pydantic also supports more advanced validation, such as regular expressions, minimum and maximum values, and custom validation functions. You can define these validations using Pydantic’s validator decorators. For example, you can ensure that the title is not empty and has a maximum length: “`python from pydantic import BaseModel, validator

class CreateBlogPost(BaseModel): title: str content: str publication_date: datetime = datetime.now()

@validator(‘title’) def title_must_not_be_empty(cls, value): if not value.strip(): raise ValueError(‘Title cannot be empty’) if len(value) > 100: raise ValueError(‘Title is too long’) return value 


## Building API Endpoints for Blog Posts

With our models defined, let's create the API endpoints for managing blog posts. We'll need endpoints for creating, reading, updating, and deleting blog posts. These endpoints will handle the HTTP requests and interact with our data models. First, let’s create the endpoint for creating new blog posts. This endpoint will accept a `CreateBlogPost` object in the request body and return the created `BlogPost` object. Import the necessary modules and models in `main.py`:

```python
from typing import List
from fastapi import FastAPI, HTTPException
from models import BlogPost, CreateBlogPost

app = FastAPI()

blog_posts = []

@app.post("/posts/", response_model=BlogPost, status_code=201)
async def create_post(post: CreateBlogPost):
 post_id = len(blog_posts) + 1
 new_post = BlogPost(id=post_id, **post.dict())
 blog_posts.append(new_post)
 return new_post

In this endpoint, @app.post("/posts/") defines a POST endpoint for creating new blog posts. response_model=BlogPost specifies that the endpoint will return a BlogPost object in the response. status_code=201 sets the HTTP status code to 201 Created, which is the standard status code for successful resource creation. The create_post function accepts a CreateBlogPost object in the request body, which is automatically validated by Pydantic. The function generates a new ID for the blog post, creates a BlogPost object, appends it to the blog_posts list, and returns the created blog post. Next, let’s create the endpoint for reading a blog post by its ID. This endpoint will accept an ID in the path and return the corresponding BlogPost object. If the blog post is not found, it will raise an HTTP 404 Not Found exception: 

@app.get("/posts/{post_id}", response_model=BlogPost)
async def read_post(post_id: int):
 for post in blog_posts:
 if post.id == post_id:
 return post
 raise HTTPException(status_code=404, detail="Post not found")

In this endpoint, @app.get("/posts/{post_id}") defines a GET endpoint for reading a blog post by its ID. {post_id} is a path parameter that accepts an integer value. response_model=BlogPost specifies that the endpoint will return a BlogPost object in the response. The read_post function accepts the post_id as an argument. It iterates through the blog_posts list and returns the blog post with the matching ID. If no matching blog post is found, it raises an HTTPException with a status code of 404 and a detail message indicating that the post was not found. Now, let’s create the endpoint for updating an existing blog post. This endpoint will accept an ID in the path and a CreateBlogPost object in the request body. It will update the corresponding blog post and return the updated BlogPost object. If the blog post is not found, it will raise an HTTP 404 Not Found exception:

See also: San Antonio Flag Football: Your Ultimate Guide 
@app.put("/posts/{post_id}", response_model=BlogPost)
async def update_post(post_id: int, post: CreateBlogPost):
 for index, existing_post in enumerate(blog_posts):
 if existing_post.id == post_id:
 updated_post = BlogPost(id=post_id, **post.dict())
 blog_posts[index] = updated_post
 return updated_post
 raise HTTPException(status_code=404, detail="Post not found")

In this endpoint, @app.put("/posts/{post_id}") defines a PUT endpoint for updating an existing blog post. response_model=BlogPost specifies that the endpoint will return a BlogPost object in the response. The update_post function accepts the post_id and a CreateBlogPost object in the request body. It iterates through the blog_posts list and updates the blog post with the matching ID. If no matching blog post is found, it raises an HTTPException with a status code of 404 and a detail message indicating that the post was not found. Finally, let’s create the endpoint for deleting a blog post. This endpoint will accept an ID in the path and return a success message. If the blog post is not found, it will raise an HTTP 404 Not Found exception: 

@app.delete("/posts/{post_id}")
async def delete_post(post_id: int):
 for index, post in enumerate(blog_posts):
 if post.id == post_id:
 del blog_posts[index]
 return {"message": "Post deleted successfully"}
 raise HTTPException(status_code=404, detail="Post not found")

In this endpoint, @app.delete("/posts/{post_id}") defines a DELETE endpoint for deleting a blog post. The delete_post function accepts the post_id as an argument. It iterates through the blog_posts list and deletes the blog post with the matching ID. If no matching blog post is found, it raises an HTTPException with a status code of 404 and a detail message indicating that the post was not found. Now that we have created the API endpoints for managing blog posts, you can test them using tools like curl or Postman. These endpoints allow you to perform CRUD (Create, Read, Update, Delete) operations on blog posts.

Securing Your FastAPI Application

Security is paramount , guys. We need to protect our API from unauthorized access. FastAPI makes it relatively straightforward to implement authentication and authorization. Let’s add some security measures to our blog API. First, let’s implement authentication using JWT (JSON Web Tokens) . JWTs are a standard way of representing claims securely between two parties. We’ll use the python-jose library to create and verify JWTs. Install it using pip: pip install python-jose . Next, we need to define a user model and authentication endpoints. Create a new file named auth.py and define a User model using Pydantic: 

from pydantic import BaseModel

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

In this model, username and password are required strings. In a real-world application, you would store the user credentials in a database and hash the passwords. However, for simplicity, we’ll use a hardcoded user for demonstration purposes. Next, let’s define the authentication endpoints in auth.py : 

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
from .models import User

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


async def authenticate_user(form_data: OAuth2PasswordRequestForm = Depends()):
 user = User(username=form_data.username, password=form_data.password)
 if user.username == "testuser" and user.password == "password":
 return user
 raise HTTPException(status_code=400, detail="Incorrect username or password")


def create_access_token(data: dict, expires_delta: timedelta | None = None):
 to_encode = data.copy()
 if expires_delta:
 expire = datetime.utcnow() + expires_delta
 else:
 expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
 to_encode.update({"exp": expire})
 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
 return encoded_jwt


async def get_current_user(token: str = Depends(oauth2_scheme)):
 credentials_exception = HTTPException(
 status_code=401,
 detail="Could not validate credentials",
 headers={"WWW-Authenticate": "Bearer"},
 )
 try:
 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
 username: str = payload.get("sub")
 if username is None:
 raise credentials_exception
 except JWTError:
 raise credentials_exception
 user = User(username=username, password="")
 return user

In this code, OAuth2PasswordBearer is used to define the token URL, which is /token in this case. The authenticate_user function checks the provided username and password against a hardcoded user. The create_access_token function creates a JWT token with an expiration time. The get_current_user function verifies the JWT token and returns the current user. Now, let’s add the authentication endpoints to main.py : 

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordRequestForm
from auth import authenticate_user, create_access_token, get_current_user
from models import User

app = FastAPI()

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
 user = await authenticate_user(form_data)
 if not user:
 raise HTTPException(status_code=400, detail="Incorrect username or password")
 access_token_expires = timedelta(minutes=30)
 access_token = create_access_token(
 data={"sub": user.username},
 expires_delta=access_token_expires,
 )
 return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
 return current_user

In this code, the /token endpoint accepts a username and password, authenticates the user, and returns an access token. The /users/me endpoint requires a valid access token and returns the current user’s information. Now that we have implemented authentication, let’s add authorization to our blog post endpoints. We’ll require a valid access token to create, update, and delete blog posts. Modify the blog post endpoints in main.py to require authentication: 

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from auth import authenticate_user, create_access_token, get_current_user
from models import User, BlogPost, CreateBlogPost

app = FastAPI()

blog_posts = []

@app.post("/posts/", response_model=BlogPost, status_code=201)
async def create_post(post: CreateBlogPost, current_user: User = Depends(get_current_user)):
 post_id = len(blog_posts) + 1
 new_post = BlogPost(id=post_id, **post.dict())
 blog_posts.append(new_post)
 return new_post


@app.put("/posts/{post_id}", response_model=BlogPost)
async def update_post(post_id: int, post: CreateBlogPost, current_user: User = Depends(get_current_user)):
 for index, existing_post in enumerate(blog_posts):
 if existing_post.id == post_id:
 updated_post = BlogPost(id=post_id, **post.dict())
 blog_posts[index] = updated_post
 return updated_post
 raise HTTPException(status_code=404, detail="Post not found")


@app.delete("/posts/{post_id}")
async def delete_post(post_id: int, current_user: User = Depends(get_current_user)):
 for index, post in enumerate(blog_posts):
 if post.id == post_id:
 del blog_posts[index]
 return {"message": "Post deleted successfully"}
 raise HTTPException(status_code=404, detail="Post not found")

In these endpoints, the current_user: User = Depends(get_current_user) parameter is added to require a valid access token. Now, only authenticated users can create, update, and delete blog posts. This ensures that your API is protected from unauthorized access. Remember to replace "your-secret-key" with a strong, randomly generated secret key in your actual application. This secret key is used to sign and verify the JWT tokens, and it should be kept confidential. Always use HTTPS in production to protect the access tokens from being intercepted.

Writing Tests for Your FastAPI Application

Testing is crucial for ensuring the reliability of your API. FastAPI works seamlessly with Pytest, a popular testing framework for Python. Let’s write some tests for our blog API. First, install Pytest and the FastAPI test client using pip: pip install pytest httpx . Next, create a new file named test_main.py and define the test cases. Here’s an example: 

import pytest
from httpx import AsyncClient
from fastapi import FastAPI
from main import app

@pytest.fixture
async def async_client() -> AsyncClient:
 async with AsyncClient(app=app, base_url="http://test") as client:
 yield client


@pytest.mark.asyncio
async def test_create_post(async_client: AsyncClient):
 response = await async_client.post("/posts/", json={"title": "Test Post", "content": "This is a test post"})
 assert response.status_code == 201
 assert response.json()["title"] == "Test Post"
 assert response.json()["content"] == "This is a test post"


@pytest.mark.asyncio
async def test_read_post(async_client: AsyncClient):
 await async_client.post("/posts/", json={"title": "Test Post", "content": "This is a test post"})
 response = await async_client.get("/posts/1")
 assert response.status_code == 200
 assert response.json()["title"] == "Test Post"
 assert response.json()["content"] == "This is a test post"

In this code, AsyncClient from httpx is used to send HTTP requests to our API in the test cases. The test_create_post function tests the creation of a new blog post by sending a POST request to the /posts/ endpoint and asserting that the response status code is 201 and the response body contains the expected data. The test_read_post function tests the retrieval of a blog post by sending a GET request to the /posts/1 endpoint and asserting that the response status code is 200 and the response body contains the expected data. You can run the tests using the command pytest . This will execute the test cases and report any failures. Writing tests helps ensure that your API functions correctly and that changes to the code do not introduce regressions. Test-driven development (TDD) is a development approach where you write tests before writing the actual code. This helps you think about the requirements and design of your API before you start implementing it. When writing tests, consider the different scenarios and edge cases that your API might encounter. For example, you should test the API with invalid data, missing data, and large amounts of data to ensure that it handles these cases gracefully. You should also test the API with different HTTP methods, such as GET, POST, PUT, and DELETE, to ensure that all endpoints are functioning correctly. Remember to keep your tests simple and focused. Each test should test a specific aspect of your API, and it should be easy to understand what the test is doing. Avoid writing complex tests that are difficult to maintain. By writing comprehensive tests, you can ensure that your API is reliable, robust, and easy to maintain.

Conclusion

Alright, awesome job guys! We’ve covered a lot in this guide. You’ve learned how to set up a FastAPI project, define data models with Pydantic, build API endpoints for managing blog posts, secure your application with JWT authentication, and write tests to ensure its reliability. FastAPI is an incredibly powerful and versatile framework, and this is just the beginning. Keep exploring, keep building, and most importantly, keep learning! Happy coding!

New Post