Utilizing top-tier programming language frameworks simplifies the process of generating high-quality products rapidly. Exceptional frameworks even make the entire development process enjoyable. FastAPI, a cutting-edge Python web framework, offers both power and an enjoyable user experience. The following aspects make exploring the Python FastAPI framework worthwhile:
- Speed: FastAPI stands out as one of the swiftest Python web frameworks. In fact, its speed is on par with Node.js and Go. Refer to these FastAPI performance tests for more information.
- The FastAPI documentation is comprehensive yet user-friendly.
- By adding type hints to your code, you gain automatic data validation and conversion.
- Dependency injection simplifies the creation of plugins.
A Practical Introduction to Python FastAPI: Constructing a TODO Application
To illustrate the core concepts of FastAPI, let’s develop a TODO application that allows users to manage to-do lists. Our example FastAPI application will include the following functionalities:
- User Signup and Login
- Adding New TODO Items
- Retrieving a List of All TODOs
- Deleting and Updating TODO Items
Utilizing SQLAlchemy to Define Data Models
Our application consists of two primary models: User and TODO. With the assistance of SQLAlchemy, a Python database toolkit, we can represent these models as follows:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
lname = Column(String)
fname = Column(String)
email = Column(String, unique=True, index=True)
todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan")
class TODO(Base):
__tablename__ = "todos"
id = Column(Integer, primary_key=True, index=True)
text = Column(String, index=True)
completed = Column(Boolean, default=False)
owner_id = Column(Integer, ForeignKey("users.id"))
owner = relationship("User", back_populates="todos")
|
With our models defined, we’ll configure SQLAlchemy to establish a database connection.
1
2
3
4
5
6
7
8
9
| import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL']
engine = create_engine(
SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
|
Harnessing the Capabilities of Type Hints
A significant portion of API development involves routine tasks such as data validation and conversion. Let’s address these upfront before delving into request handler creation. FastAPI leverages pydantic models to define the structure of incoming and outgoing data. By utilizing these models for type hinting, we benefit from automated data validation and conversion. Keep in mind that these models are independent of our database workflow and solely define the data format exchanged through our REST interface. When defining pydantic models, consider all potential data flows related to User and TODO information.
Typically, new users sign up for our TODO service, while existing users log in. Both interactions involve User information, but the data structure differs. Signup requires more comprehensive user data compared to login, which only needs email and password. Consequently, we need two pydantic models to represent these distinct User info structures.
However, our TODO app will utilize FastAPI’s built-in OAuth2 support for a JSON Web Tokens (JWT)-based login process. Therefore, we’ll define a UserCreate schema for data submitted to the signup endpoint and a UserBase schema for successful signup responses.
1
2
3
4
5
6
7
8
| from pydantic import BaseModel
from pydantic import EmailStr
class UserBase(BaseModel):
email: EmailStr
class UserCreate(UserBase):
lname: str
fname: str
password: str
|
Here, we’ve defined last name, first name, and password as strings. However, we can enforce stricter validation using pydantic constrained strings, enabling checks for minimum/maximum length and regex patterns.
To support TODO item creation and listing, we define the following schema:
1
2
3
| class TODOCreate(BaseModel):
text: str
completed: bool
|
For updating existing TODO items, we use this schema:
1
2
| class TODOUpdate(TODOCreate):
id: int
|
With our data exchange schemas defined, we can now focus on request handlers, where these schemas will automate data conversion and validation.
Enabling User Signup
First, let’s implement user signup functionality, as all our services require user authentication. Our first request handler utilizes the UserCreate and UserBase schemas defined earlier.
1
2
3
4
5
6
7
8
9
| @app.post("/api/users", response_model=schemas.User)
def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)):
"""add new user"""
user = crud.get_user_by_email(db, user_data.email)
if user:
raise HTTPException(status_code=409,
detail="Email already registered.")
signedup_user = crud.create_user(db, user_data)
return signedup_user
|
This concise code snippet accomplishes several tasks. A decorator specifies the HTTP verb, URI, and successful response schema. Type hinting the request body with the UserCreate schema ensures data submission correctness. The method utilizes dependency injection to access the database, a concept explored later in this FastAPI tutorial.
Enhancing API Security
We aim to incorporate the following security features into our FastAPI example:
- Password Hashing
- JWT-Based Authentication
For password hashing, we’ll utilize Passlib. Let’s define functions for password hashing and verification.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
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 authenticate_user(db, email: str, password: str):
user = crud.get_user_by_email(db, email)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
|
To enable JWT-based authentication, we need functions for JWT generation and decoding to retrieve user credentials.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
| # install PyJWT
import jwt
from fastapi.security import OAuth2PasswordBearer
SECRET_KEY = os.environ['SECRET_KEY']
ALGORITHM = os.environ['ALGORITHM']
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
def decode_access_token(db, token):
credentials_exception = HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise credentials_exception
token_data = schemas.TokenData(email=email)
except PyJWTError:
raise credentials_exception
user = crud.get_user_by_email(db, email=token_data.email)
if user is None:
raise credentials_exception
return user
|
Issuing Tokens upon Successful Login
Now, let’s create a Login endpoint implementing the OAuth2 password flow. This endpoint receives user email and password, verifies them against the database, and issues a JSON web token upon successful authentication.
We’ll utilize OAuth2PasswordRequestForm from FastAPI’s security utilities to handle credential submission.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| @app.post("/api/token", response_model=schemas.Token)
def login_for_access_token(db: Session = Depends(get_db),
form_data: OAuth2PasswordRequestForm = Depends()):
"""generate access token for valid credentials"""
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(data={"sub": user.email},
expires_delta=access_token_expires)
return {"access_token": access_token, "token_type": "bearer"}
|
Leveraging Dependency Injection for Database Access and Endpoint Protection
With the login endpoint set up to provide JWTs, users can store these tokens locally and include them as an Authorization header in subsequent requests. Protected endpoints can then decode these tokens to identify the requester.
This logic is not specific to any particular endpoint but is shared across all protected endpoints. Therefore, we’ll implement token decoding as a dependency reusable by any request handler.
In FastAPI terminology, our path operation functions (request handlers) will depend on get_current_user. This dependency needs database access and integration with FastAPI’s OAuth2PasswordBearer logic for token retrieval. We achieve this by making get_current_user depend on other functions, establishing a powerful dependency chain.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def get_db():
"""provide db session to path operation functions"""
try:
db = SessionLocal()
yield db
finally:
db.close()
def get_current_user(db: Session = Depends(get_db),
token: str = Depends(oauth2_scheme)):
return decode_access_token(db, token)
@app.get("/api/me", response_model=schemas.User)
def read_logged_in_user(current_user: models.User = Depends(get_current_user)):
"""return user settings for current user"""
return current_user
|
Enabling CRUD Operations for Logged-in Users
Before implementing the TODO Create, Read, Update, Delete (CRUD) path operation functions, let’s define helper functions for database-level CRUD operations.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate):
todo = models.TODO(text=todo_data.text,
completed=todo_data.completed)
todo.owner = current_user
db.add(todo)
db.commit()
db.refresh(todo)
return todo
def update_todo(db: Session, todo_data: schemas.TODOUpdate):
todo = db.query(models.TODO).filter(models.TODO.id == id).first()
todo.text = todo_data.text
todo.completed = todo.completed
db.commit()
db.refresh(todo)
return todo
def delete_todo(db: Session, id: int):
todo = db.query(models.TODO).filter(models.TODO.id == id).first()
db.delete(todo)
db.commit()
def get_user_todos(db: Session, userid: int):
return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()
|
These database functions will be utilized by the following REST endpoints:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| @app.get("/api/mytodos", response_model=List[schemas.TODO])
def get_own_todos(current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""return a list of TODOs owned by current user"""
todos = crud.get_user_todos(db, current_user.id)
return todos
@app.post("/api/todos", response_model=schemas.TODO)
def add_a_todo(todo_data: schemas.TODOCreate,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""add a TODO"""
todo = crud.create_meal(db, current_user, meal_data)
return todo
@app.put("/api/todos/{todo_id}", response_model=schemas.TODO)
def update_a_todo(todo_id: int,
todo_data: schemas.TODOUpdate,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""update and return TODO for given id"""
todo = crud.get_todo(db, todo_id)
updated_todo = crud.update_todo(db, todo_id, todo_data)
return updated_todo
@app.delete("/api/todos/{todo_id}")
def delete_a_meal(todo_id: int,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(get_db)):
"""delete TODO of given id"""
crud.delete_meal(db, todo_id)
return {"detail": "TODO Deleted"}
|
Writing Tests
Let’s write some tests for our TODO API using FastAPI’s TestClient, based on the popular Requests library, and execute them with Pytest.
To ensure only authenticated users can create TODOs, we can write a test like this:
1
2
3
4
5
6
| from starlette.testclient import TestClient
from .main import app
client = TestClient(app)
def test_unauthenticated_user_cant_create_todos(): todo=dict(text="run a mile", completed=False)
response = client.post("/api/todos", data=todo)
assert response.status_code == 401
|
The following test checks our login endpoint and generates a JWT when provided with valid credentials.
1
2
3
4
5
| def test_user_can_obtain_auth_token():
response = client.post("/api/token", data=good_credentials)
assert response.status_code == 200
assert 'access_token' in response.json()
assert 'token_type' in response.json()
|
In Conclusion
We’ve successfully implemented a basic TODO application using FastAPI, demonstrating the power of type hints in defining data structures for our REST interface. By defining schemas once, FastAPI handles data validation and conversion automatically. Another notable feature is dependency injection, used for shared logic such as database connection, JWT decoding, and implementing OAuth2 with password and bearer tokens. We also explored dependency chaining.
This approach can be easily extended to incorporate features like role-based access. FastAPI’s exceptional performance and documentation simplify learning and usage. We can write concise, powerful code without grappling with framework-specific complexities. In essence, FastAPI provides a collection of robust tools that feel like natural extensions of modern Python. Enjoy using it!