Starting Web Development with FastAPI

FastAPI is a modern, fast (high-performance) web framework for building APIs with Python 3.7+ based on standard Python type hints. It is designed to be easy to use and to offer high performance, comparable to Node.js and Go.

This comprehensive guide will take you through the steps of setting up a web development environment with FastAPI, building endpoints, handling requests and responses, integrating a database, adding authentication and authorization, and deploying your application.

Table of Contents

What is FastAPI?
Setting Up the Environment
Creating a Basic FastAPI Application
Handling Requests and Responses
Database Integration with SQLAlchemy
Authentication and Authorization
Error Handling and Validation
Testing Your API
Deploying Your Application

Conclusion

What is FastAPI?

FastAPI is a web framework for building APIs with Python 3.7+ that is fast to code and run. It leverages Python’s type hints for data validation, serialization, and documentation generation.

Key Features:

High Performance: FastAPI is built on Starlette and Pydantic, ensuring high performance and automatic validation.
Ease of Use: It is designed to be easy to use and intuitive, reducing development time.
Interactive Documentation: FastAPI generates interactive API documentation with Swagger UI and ReDoc.
Standards-based: It is based on OpenAPI and JSON Schema standards, making it highly interoperable.

Setting Up the Environment

To start developing with FastAPI, you need to set up your development environment. Follow these steps:

Install Python

Ensure you have Python installed on your system. You can download it from the official Python website.

Create a Virtual Environment

It’s good practice to create a virtual environment for your projects to manage dependencies. Run the following commands:
python -m venv venv source venv/bin/activate # On Windows, use `venv\Scripts\activate`

Install FastAPI and Uvicorn
Use pip to install FastAPI and Uvicorn (an ASGI server for running FastAPI applications):
pip install fastapi uvicorn

3. Creating a Basic FastAPI Application

Let’s start by creating a basic FastAPI application. Create a file named main.py and add the following code:
from fastapi import FastAPI


app = FastAPI()
@app.get("/")

def read_root():

    return {"Hello": "World"}
@app.get("/items/{item_id}")

def read_item(item_id: int, q: str = None):

    return {"item_id": item_id, "q": q}

if __name__ == '__main__': import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)

In this example, we define a basic FastAPI application with two endpoints: the root endpoint (“/”) and an endpoint to retrieve an item by its ID (“/items/{item_id}”).

4. Handling Requests and Responses

Handling requests and responses is a fundamental part of web development. FastAPI makes this process straightforward with its use of Python type hints.

Handling GET Requests
GET requests are used to retrieve data from the server. The / and /items/{item_id} endpoints in the example above demonstrate handling GET requests.

Handling POST Requests
POST requests are used to create new resources. Here’s how you can add an endpoint for creating items:
from pydantic import BaseModel


class Item(BaseModel):

    name: str

    description: str = None

    price: float

    tax: float = None

@app.post("/items/") def create_item(item: Item): return {"item": item}

In this example, we define an Item model using Pydantic, which FastAPI uses for data validation and serialization.

Handling PUT and DELETE Requests
You can handle PUT and DELETE requests similarly. Here’s how you can add endpoints for updating and deleting items:
@app.put("/items/{item_id}") def update_item(item_id: int, item: Item): return {"item_id": item_id, "item": item}

@app.delete("/items/{item_id}") def delete_item(item_id: int): return {"item_id": item_id}

5. Database Integration with SQLAlchemy

For more complex applications, you’ll need a database to store your data. SQLAlchemy is a popular ORM (Object-Relational Mapping) tool for Python that works well with FastAPI.

Install SQLAlchemy
Install SQLAlchemy and databases (a SQLAlchemy wrapper for asynchronous support):
pip install sqlalchemy databases[sqlite]

Set Up the Database
Modify main.py to integrate SQLAlchemy:
from sqlalchemy import create_engine, Column, Integer, String, Float from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker


DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()
Base = declarative_base()
class Item(Base):

    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)

    name = Column(String, index=True)

    description = Column(String, index=True)

    price = Column(Float, index=True)

    tax = Column(Float, index=True)
engine = create_engine(DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base.metadata.create_all(bind=engine)

Create CRUD Operations
Create a file named crud.py and add the following CRUD operations:
from sqlalchemy.orm import Session from . import models, schemas


def get_item(db: Session, item_id: int):

    return db.query(models.Item).filter(models.Item.id == item_id).first()
def get_items(db: Session, skip: int = 0, limit: int = 10):

    return db.query(models.Item).offset(skip).limit(limit).all()
def create_item(db: Session, item: schemas.ItemCreate):

    db_item = models.Item(name=item.name, description=item.description, price=item.price, tax=item.tax)

    db.add(db_item)

    db.commit()

    db.refresh(db_item)

    return db_item
def update_item(db: Session, item_id: int, item: schemas.ItemUpdate):

    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()

    db_item.name = item.name

    db_item.description = item.description

    db_item.price = item.price

    db_item.tax = item.tax

    db.commit()

    db.refresh(db_item)

    return db_item

def delete_item(db: Session, item_id: int): db_item = db.query(models.Item).filter(models.Item.id == item_id).first() db.delete(db_item) db.commit() return db_item

Update Main Application
Update main.py to use the database and CRUD operations:
from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session from . import crud, models, schemas from .database import SessionLocal, engine


models.Base.metadata.create_all(bind=engine)
app = FastAPI()
def get_db():

    db = SessionLocal()

    try:

        yield db

    finally:

        db.close()
@app.post("/items/", response_model=schemas.Item)

def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):

    return crud.create_item(db=db, item=item)
@app.get("/items/{item_id}", response_model=schemas.Item)

def read_item(item_id: int, db: Session = Depends(get_db)):

    db_item = crud.get_item(db=db, item_id=item_id)

    if db_item is None:

        raise HTTPException(status_code=404, detail="Item not found")

    return db_item
@app.put("/items/{item_id}", response_model=schemas.Item)

def update_item(item_id: int, item: schemas.ItemUpdate, db: Session = Depends(get_db)):

    return crud.update_item(db=db, item_id=item_id, item=item)

@app.delete("/items/{item_id}", response_model=schemas.Item) def delete_item(item_id: int, db: Session = Depends(get_db)): return crud.delete_item(db=db, item_id=item_id)

6. Authentication and Authorization

To secure your API, you need to implement authentication and authorization. FastAPI provides several tools to help with this, including OAuth2 and JWT.

Install Authentication Libraries
Install the necessary libraries for JWT authentication:
pip install pyjwt passlib[bcrypt]

Add Authentication
Modify main.py to include JWT authentication:
from fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext from datetime import datetime, timedelta


# to get a string like this run: openssl rand -hex 32

SECRET_KEY = "09a7ffba7346b97f6d9f5e07b2f65c3e"

ALGORITHM = "HS256"

ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_password(plain_password, hashed_password):

    return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):

    return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: timedelta = None):

    to_encode = data.copy()

    if expires_delta:

        expire = datetime.utcnow() + expires_delta

    else:

        expire = datetime.utcnow() + timedelta(minutes=15)

    to_encode.update({"exp": expire})

    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

    return encoded_jwt
@app.post("/token", response_model=Token)

def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):

    user = authenticate_user(fake_users_db, form_data.username, form_data.password)

    if not user:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="Incorrect username or password",

            headers={"WWW-Authenticate": "Bearer"},

        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)

    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)

def read_users_me(current_user: User = Depends(get_current_user)):

    return current_user

def get_current_user(token: str = Depends(oauth2_scheme)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, 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 = get_user(fake_users_db, username=username) if user is None: raise credentials_exception return user

7. Error Handling and Validation

Proper error handling and validation are crucial for building robust APIs. FastAPI provides tools to handle errors and validate data.

Error Handling
You can customize error responses by defining error handlers. Modify main.py to include error handlers:
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException) def http_exception_handler(request, exc): return JSONResponse( status_code=exc.status_code, content={"message": exc.detail}, )

Data Validation
Use Pydantic models for data validation and serialization:
from pydantic import BaseModel, Field


class ItemCreate(BaseModel):

    name: str = Field(..., max_length=50)

    description: str = Field(None, max_length=300)

    price: float = Field(..., gt=0)

    tax: float = Field(None, ge=0)

@app.post("/items/", response_model=Item) def create_item(item: ItemCreate, db: Session = Depends(get_db)): return crud.create_item(db=db, item=item)

8. Testing Your API

Testing is essential to ensure your API works as expected. Use pytest and FastAPI’s built-in testing capabilities for writing and running tests.

Install pytest
Install pytest:
pip install pytest

Write Tests
Create a file named test_main.py and add the following tests:
from fastapi.testclient import TestClient from .main import app


client = TestClient(app)
def test_read_root():

    response = client.get("/")

    assert response.status_code == 200

    assert response.json() == {"Hello": "World"}

def test_create_item(): response = client.post( "/items/", json={"name": "Test Item", "description": "A test item", "price": 10.5, "tax": 1.0}, ) assert response.status_code == 201 assert response.json()["name"] == "Test Item" assert response.json()["description"] == "A test item" assert response.json()["price"] == 10.5 assert response.json()["tax"] == 1.0

Run Tests
Run your tests using pytest:
pytest

9. Deploying Your Application

Once your application is ready, you can deploy it to a hosting service like Heroku, AWS, or DigitalOcean.

Deploy to Heroku
Install the Heroku CLI and log in:
curl https://cli-assets.heroku.com/install.sh | sh heroku login

Create a Procfile with the following content:
web: uvicorn main:app --host=0.0.0.0 --port=${PORT:-8000}

Initialize a Git repository, commit your code, and create a Heroku app:
git init heroku create

Deploy your application:
git add . git commit -m "Initial commit" git push heroku master

Open your deployed application:
heroku open

FastAPI provides a robust yet simple framework for building high-performance APIs with Python. By following the steps in this guide, you can set up your development environment, create a basic FastAPI application, handle requests and responses, integrate with a database, add authentication and authorization, handle errors and validation, test your API, and deploy your application.

FastAPI’s ease of use, performance, and interactive documentation make it a great choice for both beginners and experienced developers looking to build web services quickly and efficiently.

Tag: Starting Web Development with FastAPI