Search icon CANCEL
Arrow left icon
Explore Products
Best Sellers
New Releases
Books
Videos
Audiobooks
Learning Hub
Conferences
Free Learning
Arrow right icon
Arrow up icon
GO TO TOP
Building Python Microservices with FastAPI

You're reading from   Building Python Microservices with FastAPI Build secure, scalable, and structured Python microservices from design concepts to infrastructure

Arrow left icon
Product type Paperback
Published in Aug 2022
Publisher Packt
ISBN-13 9781803245966
Length 420 pages
Edition 1st Edition
Languages
Tools
Arrow right icon
Author (1):
Arrow left icon
Sherwin John C. Tragura Sherwin John C. Tragura
Author Profile Icon Sherwin John C. Tragura
Sherwin John C. Tragura
Arrow right icon
View More author details
Toc

Table of Contents (17) Chapters Close

Preface 1. Part 1: Application-Related Architectural Concepts for FastAPI microservice development
2. Chapter 1: Setting Up FastAPI for Starters FREE CHAPTER 3. Chapter 2: Exploring the Core Features 4. Chapter 3: Investigating Dependency Injection 5. Chapter 4: Building the Microservice Application 6. Part 2: Data-Centric and Communication-Focused Microservices Concerns and Issues
7. Chapter 5: Connecting to a Relational Database 8. Chapter 6: Using a Non-Relational Database 9. Chapter 7: Securing the REST APIs 10. Chapter 8: Creating Coroutines, Events, and Message-Driven Transactions 11. Part 3: Infrastructure-Related Issues, Numerical and Symbolic Computations, and Testing Microservices
12. Chapter 9: Utilizing Other Advanced Features 13. Chapter 10: Solving Numerical, Symbolic, and Graphical Problems 14. Chapter 11: Adding Other Microservice Features 15. Index 16. Other Books You May Enjoy

Managing user requests and server response

Clients can pass their request data to FastAPI endpoint URLs through path parameters, query parameters, or headers to pursue service transactions. There are standards and ways to use these parameters to obtain incoming requests. Depending on the goal of the services, we use these parameters to influence and build the necessary responses the clients need. But before we discuss these various parameter types, let us explore first how we use type hinting in FastAPI’s local parameter declaration.

Parameter type declaration

All request parameters are required to be type-declared in the method signature of the service method applying the PEP 484 standard called type hints. FastAPI supports common types such as None, bool, int, and float and container types such as list, tuple, dict, set, frozenset, and deque. Other complex Python types such as datetime.date, datetime.time, datetime.datetime, datetime.delta, UUID, bytes, and Decimal are also supported.

The framework also supports the data types included in Python’s typing module, responsible for type hints. These data types are standard notations for Python and variable type annotations that can help to pursue type checking and model validation during compilation, such as Optional, List, Dict, Set, Union, Tuple, FrozenSet, Iterable, and Deque.

Path parameters

FastAPI allows you to obtain request data from the endpoint URL of an API through a path parameter or path variable that makes the URL somewhat dynamic. This parameter holds a value that becomes part of a URL indicated by curly braces ({}). After setting off these path parameters within the URL, FastAPI requires these parameters to be declared by applying type hints.

The following delete_user() service is a DELETE API method that uses a username path parameter to search for a login record for deletion:

@app.delete("/ch01/login/remove/{username}")
def delete_user(username: str):
    if username == None:
    return {"message": "invalid user"}
else:
    del valid_users[username]
    return {"message": "deleted user"}

Multiple path parameters are acceptable if the leftmost variables are more likely to be filled with values than the rightmost variables. In other words, the importance of the leftmost path variables will make the process more relevant and correct than those on the right. This standard is applied to ensure that the endpoint URL will not look like other URLs, which might cause some conflicts and confusion. The following login_with_token() service follows this standard, since username is a primary key and is as strong as, or even stronger than, its next parameter, password. There is an assurance that the URL will always look unique every time the endpoint is accessed because username will always be required, as well as password:

@app.get("/ch01/login/{username}/{password}")
def login_with_token(username: str, password:str, 
                     id: UUID):
    if valid_users.get(username) == None:
        return {"message": "user does not exist"}
    else:
        user = valid_users[username]
        if user.id == id and checkpw(password.encode(), 
                 user.passphrase):
            return user
        else:
            return {"message": "invalid user"}

Unlike other web frameworks, FastAPI is not friendly with endpoint URLs that belong to base paths or top-level domain paths with different subdirectories. This occurrence happens when we have dynamic URL patterns that look the same as the other fixed endpoint URLs when assigned a specific path variable. These fixed URLs are implemented sequentially after these dynamic URLs. An example of these are the following services:

@app.get("/ch01/login/{username}/{password}")
def login_with_token(username: str, password:str, 
                     id: UUID):
    if valid_users.get(username) == None:
        return {"message": "user does not exist"}
    else:
        user = valid_users[username]
        if user.id == id and checkpw(password.encode(), 
                      user.passphrase.encode()):
            return user
        else:
            return {"message": "invalid user"}
@app.get("/ch01/login/details/info")
def login_info():
        return {"message": "username and password are 
                            needed"}

This will give us an HTTP Status Code 422 (Unprocessable Entity) when accessing http://localhost:8080/ch01/login/details/info. There should be no problem accessing the URL, since the API service is almost a stub or trivial JSON data. What happened in this scenario is that the fixed path’s details and info path directories were treated as username and password parameter values, respectively. Because of confusion, the built-in data validation of FastAPI will show us a JSON-formatted error message that says, {"detail":[{"loc":["query","id"],"msg":"field required","type":"value_error.missing"}]}. To fix this problem, all fixed paths should be declared first before the dynamic endpoint URLs with path parameters. Thus, the preceding login_info() service should be declared first before login_with_token().

Query parameters

A query parameter is a key–value pair supplied after the end of an endpoint URL, indicated by a question mark (?). Just like the path parameter, this also holds the request data. An API service can manage a series of query parameters separated by an ampersand (&). Like in path parameters, all query parameters are also declared in the service method. The following login service is a perfect specimen that uses query parameters:

@app.get("/ch01/login/")
def login(username: str, password: str):
    if valid_users.get(username) == None:
        return {"message": "user does not exist"}
    else:
        user = valid_users.get(username)
        if checkpw(password.encode(), 
               user.passphrase.encode()):
            return user
        else:
            return {"message": "invalid user"}

The login service method uses username and password as query parameters in the str types. Both are required parameters, and assigning them with None as parameter values will give a compiler error.

FastAPI supports query parameters that are complex types, such as list and dict. But these Python collection types cannot specify the type of objects to store unless we apply the generic type hints for Python collections. The following delete_users() and update_profile_names() APIs use generic type hints, List and Dict, in declaring query parameters that are container types with type checking and data validation:

from typing import Optional, List, Dict
@app.delete("/ch01/login/remove/all")
def delete_users(usernames: List[str]):
    for user in usernames:
        del valid_users[user]
    return {"message": "deleted users"}
@app.patch("/ch01/account/profile/update/names/{username}")
def update_profile_names(username: str, id: UUID, 
                         new_names: Dict[str, str]):
    if valid_users.get(username) == None:
        return {"message": "user does not exist"}
    elif new_names == None:
        return {"message": "new names are required"}
    else:
        user = valid_users.get(username)
        if user.id == id:
            profile = valid_profiles[username]
            profile.firstname = new_names['fname']
            profile.lastname = new_names['lname']
            profile.middle_initial = new_names['mi']
            valid_profiles[username] = profile
            return {"message": "successfully updated"}
        else:
            return {"message": "user does not exist"}

FastAPI also allows you to explicitly assign default values to service function parameters.

Default parameters

There are times that we need to specify default values to the query parameter(s) and path parameter(s) of some API services to avoid validation error messages such as field required and value_error.missing. Setting default values to parameters will allow the execution of an API method with or without supplying the parameter values. Depending on the requirement, assigned default values are usually 0 for numeric types, False for bool types, empty string for string types, an empty list ([]) for List types, and an empty dictionary ({}) for Dict types. The following delete pending users() and change_password() services show us how to apply default values to the query parameter(s) and path parameter(s):

@app.delete("/ch01/delete/users/pending")
def delete_pending_users(accounts: List[str] = []):
    for user in accounts:
        del pending_users[user]
    return {"message": "deleted pending users"}
@app.get("/ch01/login/password/change")
def change_password(username: str, old_passw: str = '',
                         new_passw: str = ''):
    passwd_len = 8
    if valid_users.get(username) == None:
        return {"message": "user does not exist"}
    elif old_passw == '' or new_passw == '':
        characters = ascii_lowercase
        temporary_passwd = 
             ''.join(random.choice(characters) for i in 
                     range(passwd_len))
        user = valid_users.get(username)
        user.password = temporary_passwd
        user.passphrase = 
                  hashpw(temporary_passwd.encode(),gensalt())
        return user
    else:
        user = valid_users.get(username)
        if user.password == old_passw:
            user.password = new_passw
            user.passphrase = hashpw(new_pass.encode(),gensalt())
            return user
        else:
            return {"message": "invalid user"}

delete_pending_users() can be executed even without passing any accounts argument, since accounts will be always an empty List by default. Likewise, change_password() can still continue its process without passing any old_passwd and new_passw, since they are both always defaulted to empty str. hashpw() is a bcrypt utility function that generates a hashed passphrase from an autogenerated salt.

Optional parameters

If the path and/or query parameter(s) of a service is/are not necessarily needed to be supplied by the user, meaning the API transactions can proceed with or without their inclusion in the request transaction, then we set them as optional. To declare an optional parameter, we need to import the Optional type from the typing module and then use it to set the parameter. It should wrap the supposed data type of the parameter using brackets ([]) and can have any default value if needed. Assigning the Optional parameter to a None value indicates that its exclusion from the parameter passing is allowed by the service, but it will hold a None value. The following services depict the use of optional parameters:

from typing import Optional, List, Dict
@app.post("/ch01/login/username/unlock")
def unlock_username(id: Optional[UUID] = None):
    if id == None:
        return {"message": "token needed"}
    else:
        for key, val in valid_users.items():
            if val.id == id:
                return {"username": val.username}
        return {"message": "user does not exist"}
@app.post("/ch01/login/password/unlock")
def unlock_password(username: Optional[str] = None, 
                    id: Optional[UUID] = None):
    if username == None:
        return {"message": "username is required"}
    elif valid_users.get(username) == None:
        return {"message": "user does not exist"}
    else:
        if id == None:
            return {"message": "token needed"}
        else:
            user = valid_users.get(username)
            if user.id == id:
                return {"password": user.password}
            else:
                return {"message": "invalid token"}

In the online academic discussion forum application, we have services such as the preceding unlock_username() and unlock_password() services that declare all their parameters as optional. Just do not forget to apply exception handling or defensive validation in your implementation when dealing with these kinds of parameters to avoid HTTP Status 500 (Internal Server Error).

Important note

The FastAPI framework does not allow you to directly assign the None value to a parameter just to declare an optional parameter. Although this is allowed with the old Python behavior, this is no longer recommended in the current Python versions for the purpose of built-in type checking and model validation.

Mixing all types of parameters

If you are planning to implement an API service method that declares optional, required, and default query and path parameters altogether, you can pursue it because the framework supports it, but approach it with some caution due to some standards and rules:

@app.patch("/ch01/account/profile/update/names/{username}")
def update_profile_names(id: UUID, username: str = '' , 
           new_names: Optional[Dict[str, str]] = None):
    if valid_users.get(username) == None:
        return {"message": "user does not exist"}
    elif new_names == None:
        return {"message": "new names are required"}
    else:
        user = valid_users.get(username)
        if user.id == id:
            profile = valid_profiles[username]
            profile.firstname = new_names['fname']
            profile.lastname = new_names['lname']
            profile.middle_initial = new_names['mi']
            valid_profiles[username] = profile
            return {"message": "successfully updated"}
        else:
            return {"message": "user does not exist"}

The updated version of the preceding update_profile_names() service declares a username path parameter, a UUID id query parameter, and an optional Dict[str, str] type. With mixed parameter types, all required parameters should be declared first, followed by default parameters, and last in the parameter list should be the optional types. Disregarding this ordering rule will generate a compiler error.

Request body

A request body is a body of data in bytes transmitted from a client to a server through a POST, PUT, DELETE, or PATCH HTTP method operation. In FastAPI, a service must declare a model object to represent and capture this request body to be processed for further results.

To implement a model class for the request body, you should first import the BaseModel class from the pydantic module. Then, create a subclass of it to utilize all the properties and behavior needed by the path operation in capturing the request body. Here are some of the data models used by our application:

from pydantic import BaseModel
class User(BaseModel):
    username: str
    password: str
class UserProfile(BaseModel):
    firstname: str
    lastname: str
    middle_initial: str
    age: Optional[int] = 0
    salary: Optional[int] = 0
    birthday: date
    user_type: UserType

The attributes of the model classes must be explicitly declared by applying type hints and utilizing the common and complex data types used in the parameter declaration. These attributes can also be set as required, default, and optional, just like in the parameters.

Moreover, the pydantic module allows the creation of nested models, even the deeply nested ones. A sample of these is shown here:

class ForumPost(BaseModel):
    id: UUID
    topic: Optional[str] = None
    message: str
    post_type: PostType
    date_posted: datetime
    username: str
class ForumDiscussion(BaseModel):
    id: UUID
    main_post: ForumPost
    replies: Optional[List[ForumPost]] = None
    author: UserProfile

As seen in the preceding code, we have a ForumPost model, which has a PostType model attribute, and ForumDiscussion, which has a List attribute of ForumPost, a ForumPost model attribute, and a UserProfile attribute. This kind of model blueprint is called a nested model approach.

After creating these model classes, you can now inject these objects into the services that are intended to capture the request body from the clients. The following services utilize our User and UserProfile model classes to manage the request body:

@app.post("/ch01/login/validate", response_model=ValidUser)
def approve_user(user: User):
    if not valid_users.get(user.username) == None:
        return ValidUser(id=None, username = None, 
             password = None, passphrase = None)
    else:
        valid_user = ValidUser(id=uuid1(), 
             username= user.username, 
             password  = user.password, 
             passphrase = hashpw(user.password.encode(),
                          gensalt()))
        valid_users[user.username] = valid_user
        del pending_users[user.username]
        return valid_user
@app.put("/ch01/account/profile/update/{username}")
def update_profile(username: str, id: UUID, 
                   new_profile: UserProfile):
    if valid_users.get(username) == None:
        return {"message": "user does not exist"}
    else:
        user = valid_users.get(username)
        if user.id == id:
            valid_profiles[username] = new_profile
            return {"message": "successfully updated"}
        else:
            return {"message": "user does not exist"}

Models can be declared required, with a default instance value, or optional in the service method, depending on the specification of the API. Missing or incorrect details such as invalid password or None values in the approve_user() service will emit the Status Code 500 (Internal Server Error). How FastAPI handles exceptions will be part of Chapter 2, Exploring the Core Features, discussions.

Important note

There are two essential points we need to emphasize when dealing with BaseModel class types. First, the pydantic module has a built-in JSON encoder that converts the JSON-formatted request body to the BaseModel object. So, there is no need create a custom converter to map the request body to the BaseModel model. Second, to instantiate a BaseModel class, all its required attributes must be initialized immediately through the constructor’s named parameters.

Request headers

In a request-response transaction, it is not only the parameters that are accessible by the REST API methods but also the information that describes the context of the client where the request originated. Some common request headers such as User-Agent, Host, Accept, Accept-Language, Accept-Encoding, Referer, and Connection usually appear with request parameters and values during request transactions.

To access a request header, import first the Header function from the fastapi module. Then, declare the variable that has the same name as the header in the method service as str types and initialize the variable by calling the Header(None) function. The None argument enables the Header() function to declare the variable optionally, which is a best practice. For hyphenated request header names, the hyphen (-) should be converted to an underscore (_); otherwise, the Python compiler will flag a syntax error message. It is the task of the Header() function to convert the underscore (_) to a hyphen (-) during request header processing.

Our online academic discussion forum application has a verify_headers() service that retrieves core request headers needed to verify a client’s access to the application:

from fastapi import Header
@app.get("/ch01/headers/verify")
def verify_headers(host: Optional[str] = Header(None), 
                   accept: Optional[str] = Header(None),
                   accept_language: 
                       Optional[str] = Header(None),
                   accept_encoding: 
                       Optional[str] = Header(None),
                   user_agent: 
                       Optional[str] = Header(None)):
    request_headers["Host"] = host
    request_headers["Accept"] = accept
    request_headers["Accept-Language"] = accept_language
    request_headers["Accept-Encoding"] = accept_encoding
    request_headers["User-Agent"] = user_agent
    return request_headers

Important note

Non-inclusion of the Header() function call in the declaration will let FastAPI treat the variables as query parameters. Be cautious also with the spelling of the local parameter names, since they are the request header names per se except for the underscore.

Response data

All API services in FastAPI should return JSON data, or it will be invalid and may return None by default. These responses can be formed using dict, BaseModel, or JSONResponse objects. Discussions on JSONResponse will be discussed in the succeeding chapters.

The pydantic module’s built-in JSON converter will manage the conversion of these custom responses to a JSON object, so there is no need to create a custom JSON encoder:

@app.post("/ch01/discussion/posts/add/{username}")
def post_discussion(username: str, post: Post, 
                    post_type: PostType):
    if valid_users.get(username) == None:
        return {"message": "user does not exist"}
    elif not (discussion_posts.get(id) == None):
        return {"message": "post already exists"}
    else:
        forum_post = ForumPost(id=uuid1(), 
          topic=post.topic, message=post.message, 
          post_type=post_type, 
          date_posted=post.date_posted, username=username)
        user = valid_profiles[username]
        forum = ForumDiscussion(id=uuid1(), 
         main_post=forum_post, author=user, replies=list())
        discussion_posts[forum.id] = forum
        return forum

The preceding post_discussion() service returns two different hardcoded dict objects, with message as the key and an instantiated ForumDiscussion model.

On the other hand, this framework allows us to specify the return type of a service method. The setting of the return type happens in the response_model attribute of any of the @app path operations. Unfortunately, the parameter only recognizes BaseModel class types:

@app.post("/ch01/login/validate", response_model=ValidUser)
def approve_user(user: User):
    
    if not valid_users.get(user.username) == None:
        return ValidUser(id=None, username = None, 
                   password = None, passphrase = None)
    else:
        valid_user = ValidUser(id=uuid1(), 
         username= user.username, password = user.password,
          passphrase = hashpw(user.password.encode(),
                 gensalt()))
        valid_users[user.username] = valid_user
        del pending_users[user.username]
        return valid_user

The preceding approve_user() service specifies the required return of the API method, which is ValidUser.

Now, let us explore how FastAPI handles form parameters.

You have been reading a chapter from
Building Python Microservices with FastAPI
Published in: Aug 2022
Publisher: Packt
ISBN-13: 9781803245966
Register for a free Packt account to unlock a world of extra content!
A free Packt account unlocks extra newsletters, articles, discounted offers, and much more. Start advancing your knowledge today.
Unlock this book and the full library FREE for 7 days
Get unlimited access to 7000+ expert-authored eBooks and videos courses covering every tech area you can think of
Renews at $19.99/month. Cancel anytime