Chapter 4. Developing the FastAPI Code
In Chapter 3, you created your database and the Python code to access the database. In this chapter, you build on this foundation code to create a working API. Table 4-1 lists the endpoints that you will create to fulfill these user stories:
Endpoint description | HTTP verb | URL |
---|---|---|
API health check |
GET |
/ |
Read player list |
GET |
/v0/players/ |
Read individual player |
GET |
/v0/players/{player_id}/ |
Read performance list |
GET |
/v0/performances/ |
Read league list |
GET |
/v0/leagues/ |
Read individual league |
GET |
/v0/leagues/{league_id}/ |
Read team list |
GET |
/v0/teams/ |
You are using version zero for your API. This will notify API consumers that the product is changing rapidly and they should be aware of potential breaking changes — changes that cause functionality to stop working, and may require consumers to make changes in their program code.
Continuing Your Portfolio Project
Figure 4-1 shows the same API components you saw previously, with one addition: the Uvicorn web server. Uvicorn will execute your API code and interact with API requests.
In Chapter 3, you completed two very important parts of the API: the SQLite database and the SQLAlchemy classes that enable Python to interact with the data. In this chapter, you will finish the rest of the components. You will create Pydantic schemas which define the structure of request and response messages. Then you will create the controlling FastAPI application, that stitches all the other components together to finish the API.
Software Used In This Chapter
The software introduced in this chapter will focus on handling API requests from your consumers. Table 4-2 displays the new tools you will use:
Software name | Version | Purpose |
---|---|---|
FastAPI |
0.111 |
Web framework to build the API |
FastAPI CLI |
0.0.4 |
Command line interface for FastAPI |
HTTPX |
0.26 |
HTTP client for Python |
Pydantic |
2.4 |
Validation library |
Uvicorn |
0.23 |
Web server to run the API |
FastAPI
FastAPI is a Python web framework that is designed for building APIs. A web framework is a set of libraries that simplify common tasks for web applications. Other common web frameworks include Express, Flask, Django, and Ruby on Rails.
FastAPI is built to be fast in both application performance and developer productivity. Because FastAPI focuses on API development, it simplifies several tasks related to API building and publishing:
-
It handles HTTP traffic, requests/responses, and other “plumbing” jobs with a few lines of code.
-
It automatically generates an OpenAPI specification file for your API, which is useful for integrating with other products.
-
It includes interactive documentation for your API.
-
It supports API versioning, security, and many other capabilities.
As you will see as you work through the portfolio project, all of these capabilities provide benefits to the users of your APIs.
Compared to the other frameworks I mentioned, FastAPI is a relative newcomer. It is an open-source project created by Sebastián Ramírez Montaño in 2018.
As of version 0.111.0, FastAPI also includes the FastAPI CLI. This is a separate Python library that is used to run FastAPI from the command line.
For your project, you will use version 0.111 of FastAPI. That version number is important because according to semantic versioning, the 0.x indicates that breaking changes may occur with the software.
HTTPX
HTTPX is a Python HTTP client. It is similar to the very popular requests
library, but it supports asynchronous calls, which allows some tasks to finish while others process. requests
only supports synchronous calls, which wait until they receive a response before continuing. HTTPX is used by Pytest to test FastAPI programs. You will also use this library in Chapter 7 to create your Python SDK.
Pydantic
Pydantic is a data validation library, which will play a key part in the APIs that you build. Because APIs are used to communicate between systems, a critical piece of their functionality is the validation of inputs and outputs. API developers and data scientists typically spend a significant amount of time writing the code to check the data types and validate values that go into and out of the API endpoints.
Pydantic is purpose-built to address this important task. Similar to FastAPI, Pydantic is fast in two ways: it saves the developer time that would be spent to write custom Python validation code, and Pydantic validation code runs much faster because it is implemented in the Rust programming language.
In addition to these benefits, objects defined in Pydantic automatically support tooltips and hints in IDEs such as Visual Studio Code. FastAPI uses Pydantic to generate JSON Schema representations from Python code. JSON Schema is a standard that ensures consistency in JSON data structures. This Pydantic feature enables FastAPI to automatically generate the OpenAPI specification, which is an industry-standard file describing APIs.
For your project, you will use Pydantic version 2.4.
Uvicorn
All web applications, including APIs, rely on a web server to handle the various administrative tasks related to handling requests and responses. You will be using the open-source Uvicorn web server. Uvicorn is based on the ASGI specification, which provides support for both synchronous processes (which block the process while waiting for a task to be performed) and asychronous processes (which can allow another process to continue while they are waiting).
For your project, you will be using Uvicorn 0.23.
Copying Files From Chapter 3
To continue your portfolio project where you left it in the previous chapter, change the directory to chapter4 and then copy the previous chapter’s files over to it. The following shows the commands and expected output:
.../portfolio-project
(
main)
$
cd
chapter4
.../chapter4
(
main)
$
cp
../chapter3/*.py
.
.../chapter4
(
main)
$
cp
../chapter3/fantasy_data.db
.
.../chapter4
(
main)
$
cp
../chapter3/requirements.txt
.
.../chapter4
(
main)
$
ls
*.*
crud.py
database.py
fantasy_data.db
models.py
readme.md
requirements.txt
test_crud.py
Installing the New Libraries in Your Codespace
In the previous chapter, you created the requirements.txt file and specified libraries to install using the pip3
package manager in Python. You will now use this process to install Pydantic, FastAPI, and Uvicorn.
Update the requirements.txt to match the following:
#Chapter 4 pip requirements SQLAlchemy>=2.0.0,<2.1.0 pydantic>=2.4.0,<2.5.0 fastapi>=0.111.0,<0.112.0 uvicorn>=0.23.0,<0.24.0 Pytest>=8.1.0,<8.2.0 httpx>=0.27.0,<0.28.0
Execute the following command to install the new libraries in your Codespace and verify that the libraries installed in the previous chapter still exist:
pip3 install -r requirements.txt
You should see a message that states that these libraries were successfully installed, such as the following:
Installingcollected
packages:
uvicorn,
pydantic,
httpx,
fastapi
Successfully
installed
fastapi-0.111.0
httpx-0.26.0
pydantic-2.4.2
uvicorn-0.23.2
Creating Python Files for Your API
You will be creating two new Python files, which are detailed in Table 4-3:
File name | Purpose |
---|---|
main.py |
FastAPI file that defines routes and controls API |
schemas.py |
Defines the Pydantic classes that validate data sent to the API |
test_main.py |
Pytest file for the FastAPI program |
Creating Pydantic Schemas
The Pydantic classes define the structure of the data that the consumer will receive in their API responses. This uses a software design pattern called data transfer objects (DTO), in which you define a format for transferring data between a producer and consumer, without the consumer needing to know the backend format. In your portfolio project, the backend and frontend classes won’t look significantly different, but using DTOs allows complete flexibility on this point.
Although you define the classes using Python code and your code interacts with them as fully formed Python objects, the consumer will receive them in an HTTP request as a JSON object. FastAPI uses Pydantic to perform the serialization process, which is converting the Python objects into JSON for the API response. This means you do not need to manage serialization in your Python code, which simplifies your program. Pydantic 2 is written in Rust and performs this task much faster than Python could. In addition to performing this de-serialization task, Python also defines the response format in the openapi.json file. This is a standard contract that uses OpenAPI and JSON Schema. This will provide multiple benefits for the consumer, as you will see in subsequent chapters. Pydantic will take data from SQLAlchemy classes and provide it to the API users.
Note
Both SQLAlchemy and Pydantic documentation refer to their classes as models, which may be confusing at times. This is extra confusing for data science work, where models have additional meanings. For clarity, this book will refer to Pydantic schemas and SQLAlchemy models.
Create a file with the following contents named schemas.py:
"""Pydantic schemas"""
from
pydantic
import
BaseModel
,
ConfigDict
from
typing
import
List
from
datetime
import
date
class
Performance
(
BaseModel
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
performance_id
:
int
player_id
:
int
week_number
:
str
fantasy_points
:
float
last_changed_date
:
date
class
PlayerBase
(
BaseModel
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
player_id
:
int
gsis_id
:
str
first_name
:
str
last_name
:
str
position
:
str
last_changed_date
:
date
class
Player
(
PlayerBase
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
performances
:
List
[
Performance
]
=
[]
class
TeamBase
(
BaseModel
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
league_id
:
int
team_id
:
int
team_name
:
str
last_changed_date
:
date
class
Team
(
TeamBase
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
players
:
List
[
PlayerBase
]
=
[]
class
League
(
BaseModel
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
league_id
:
int
league_name
:
str
scoring_type
:
str
last_changed_date
:
date
teams
:
List
[
TeamBase
]
=
[]
class
Counts
(
BaseModel
):
league_count
:
int
team_count
:
int
player_count
:
int
The schemas in this file will be used to form the responses to the API endpoints that you will define next. The primary schemas are directly returned to the endpoints and the secondary schemas are returned as an attribute of the primary schema. For example, the /v0/players/ endpoint returns a list of Player
objects (primary), which has an attribute Player.performances
(secondary). Table 4-4 shows the mapping between API endpoints and schemas:
Endpoint URL | Primary schema | Secondary schema |
---|---|---|
/ |
None |
None |
/v0/players/ |
Player |
Performance |
/v0/players/{player_id}/ |
Player |
Performance |
/v0/performances/ |
Performance |
None |
/v0/leagues/ |
League |
TeamBase |
/v0/leagues/{league_id} |
League |
TeamBase |
/v0/teams/ |
Team |
PlayerBase |
/v0/counts/ |
Counts |
None |
The Performance
class is the first and simplest schema:
class
Performance
(
BaseModel
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
performance_id
:
int
player_id
:
int
week_number
:
str
fantasy_points
:
float
last_changed_date
:
date
This class represents the scoring data that the consumer will receive. From their perspective, a performance is what happens when a player plays in a single week. If you compare the elements of this class to the SQLAlchemy models, you will see that it contains all of the elements that the Performance
model contains.
Performance
is a subclass of the Pydantic BaseModel
class, which provides a lot of built-in capabilities, including validating the data types, converting the Python object to JSON (serializing), raising intelligent errors, and connecting automatically to the SQLAlchemy models.
Tip
Notice that the Pydantic data types of individual class elements are assigned with a colon, instead of an equals sign that is used by SQLAlchemy. (This will trip you up if you’re not careful.)
The player data is represented in two schemas: PlayerBase
and Player
. Breaking the data into two classes allows you to share a limited version of the data in some situations and a full version in others. Here are those two schemas:
class
PlayerBase
(
BaseModel
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
player_id
:
int
gsis_id
:
str
first_name
:
str
last_name
:
str
position
:
str
last_changed_date
:
date
class
Player
(
PlayerBase
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
performances
:
List
[
Performance
]
=
[]
The performance data had a single Performance
schema, but the player data has two schemas. PlayerBase
is a sub-class of BaseModel
, and it has all the player fields except one: the Performance
list. Table 4-4 shows that PlayerBase
will be used as a secondary schema for the /v0/teams/ endpoint. The reason is simple: reducing the amount of data transmitted in the API call. When the API user retries a list of Team
schemas, they want to see all the players on that team without also getting a list of all the scoring performances for all the players.
The full Player
schema is a sub-class of PlayerBase
and adds the list of Performance
objects. This schema is used directly in the /v0/players/ and /v0/players/{player_id}/ endpoints. In those situations, the API user wants a list of scoring performances with the players.
To see the secondary use of PlayerBase
, examine the next two schemas:
class
TeamBase
(
BaseModel
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
league_id
:
int
team_id
:
int
team_name
:
str
last_changed_date
:
date
class
Team
(
TeamBase
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
players
:
List
[
PlayerBase
]
=
[]
The Team
object contains the statement players: List[PlayerBase] = []
. As mentioned previously, this means the items in the Team.players
are the more limited PlayerBase
schema. This is the secondary usage of PlayerBase
shown in Table 4-4 in the /v0/teams/ endpoint.
The next class is the League
schema:
class
League
(
BaseModel
):
model_config
=
ConfigDict
(
from_attributes
=
True
)
league_id
:
int
league_name
:
str
scoring_type
:
str
last_changed_date
:
date
teams
:
List
[
TeamBase
]
=
[]
By now you probably noticed that League.teams
contains TeamBase
objects. This is the secondary use of TeamBase
used in the /v0/leagues/ endpoint.
Finally, you will create a special-purpose schema to support the analytics provided by the v0/counts/ endpoint. This schema does not directly map to a database table, so it does not include the model_config
element. The name of the schema is Counts
and it includes the number of league, team, and player records in the API:
class
Counts
(
BaseModel
):
league_count
:
int
team_count
:
int
player_count
:
int
At this point, you have designed the DTOs that will be used to send data to the API consumer. You are ready for the final piece: the FastAPI controller class.
Creating Your FastAPI Controller
Now that all of the pieces are in place in the other Python files, you can tie them together with the FastAPI functionality in main.py. You can accomplish a lot with only a few lines of FastAPI code.
Create the file with the following contents named main.py:
"""FastAPI program - Chapter 4"""
from
fastapi
import
Depends
,
FastAPI
,
HTTPException
from
sqlalchemy.orm
import
Session
from
datetime
import
date
import
crud
,
schemas
from
database
import
SessionLocal
app
=
FastAPI
()
# Dependency
def
get_db
():
db
=
SessionLocal
()
try
:
yield
db
finally
:
db
.
close
()
@app
.
get
(
"/"
)
async
def
root
():
return
{
"message"
:
"API health check successful"
}
@app
.
get
(
"/v0/players/"
,
response_model
=
list
[
schemas
.
Player
])
def
read_players
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
=
None
,
first_name
:
str
=
None
,
last_name
:
str
=
None
,
db
:
Session
=
Depends
(
get_db
)):
players
=
crud
.
get_players
(
db
,
skip
=
skip
,
limit
=
limit
,
min_last_changed_date
=
minimum_last_changed_date
,
first_name
=
first_name
,
last_name
=
last_name
)
return
players
@app
.
get
(
"/v0/players/
{player_id}
"
,
response_model
=
schemas
.
Player
)
def
read_player
(
player_id
:
int
,
db
:
Session
=
Depends
(
get_db
)):
player
=
crud
.
get_player
(
db
,
player_id
=
player_id
)
if
player
is
None
:
raise
HTTPException
(
status_code
=
404
,
detail
=
"Player not found"
)
return
player
@app
.
get
(
"/v0/performances/"
,
response_model
=
list
[
schemas
.
Performance
])
def
read_performances
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
=
None
,
db
:
Session
=
Depends
(
get_db
)):
performances
=
crud
.
get_performances
(
db
,
skip
=
skip
,
limit
=
limit
,
min_last_changed_date
=
minimum_last_changed_date
)
return
performances
@app
.
get
(
"/v0/leagues/
{league_id}
"
,
response_model
=
schemas
.
League
)
def
read_league
(
league_id
:
int
,
db
:
Session
=
Depends
(
get_db
)):
league
=
crud
.
get_league
(
db
,
league_id
=
league_id
)
if
league
is
None
:
raise
HTTPException
(
status_code
=
404
,
detail
=
"League not found"
)
return
league
@app
.
get
(
"/v0/leagues/"
,
response_model
=
list
[
schemas
.
League
])
def
read_leagues
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
=
None
,
league_name
:
str
=
None
,
db
:
Session
=
Depends
(
get_db
)):
leagues
=
crud
.
get_leagues
(
db
,
skip
=
skip
,
limit
=
limit
,
min_last_changed_date
=
minimum_last_changed_date
,
league_name
=
league_name
)
return
leagues
@app
.
get
(
"/v0/teams/"
,
response_model
=
list
[
schemas
.
Team
])
def
read_teams
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
=
None
,
team_name
:
str
=
None
,
league_id
:
int
=
None
,
db
:
Session
=
Depends
(
get_db
)):
teams
=
crud
.
get_teams
(
db
,
skip
=
skip
,
limit
=
limit
,
min_last_changed_date
=
minimum_last_changed_date
,
team_name
=
team_name
,
league_id
=
league_id
)
return
teams
@app
.
get
(
"/v0/counts/"
,
response_model
=
schemas
.
Counts
)
def
get_count
(
db
:
Session
=
Depends
(
get_db
)):
counts
=
schemas
.
Counts
(
league_count
=
crud
.
get_league_count
(
db
),
team_count
=
crud
.
get_team_count
(
db
),
player_count
=
crud
.
get_player_count
(
db
))
return
counts
Let’s walk through the code in your FastAPI file. Begin with the imports:
from
fastapi
import
Depends
,
FastAPI
,
HTTPException
from
sqlalchemy
.
orm
import
Session
from
datetime
import
date
import
crud
,
schemas
from
database
import
SessionLocal
These are methods from the FastAPI library. You will use these to identify this program as a FastAPI application.
The SQLAlchemy
Session
will be used when this program calls crud.py.You will use the
date
type to query by last changed date.These imports allow the FastAPI application to reference the SQLAlchemy and Pydantic classes.
This retrieves the shared
SessionLocal
class that is used to connect to your SQLite database.
Continue reviewing the code:
app
=
FastAPI
()
# Dependency
def
get_db
():
db
=
SessionLocal
()
try
:
yield
db
finally
:
db
.
close
()
In FastAPI, the primary class you will work with is a FastAPI
class. This class by default includes the functionality to handle much of the work that an API needs to perform, without requiring you to specify every detail. You create a FastAPI
instance and name it app
. This will be used in the rest of main.py. When you execute your API from the command line using Uvicorn, you will reference main:app
referring to the app
object in the main.py.
You define the get_db()
function to create a database session and close the session when you are done with it. This function is used as a dependency in the API routes within main.py.
@app
.
get
(
"/"
)
async
def
root
():
return
{
"message"
:
"API health check successful"
}
The next command is @app.get("/")
, which is a decorator. A decorator is a statement that is added above a function definition, to give special attributes to it. In this case, the decorator defines that the async def root()
function definition will be a FastAPI request handler.
This function will be called when a consumer accesses the root URL of the API, which is equivalent to /
. It will serve as a health check for the entire API by returning a simple message to the consumer. The next statement defines the first endpoint that we have created for your user stories:
@app
.
get
(
"/v0/players/"
,
response_model
=
list
[
schemas
.
Player
])
def
read_players
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
|
None
=
None
,
first_name
:
str
|
None
=
None
,
last_name
:
str
|
None
=
None
,
db
:
Session
=
Depends
(
get_db
)):
players
=
crud
.
get_players
(
db
,
skip
=
skip
,
limit
=
limit
,
min_last_changed_date
=
minimum_last_changed_date
,
first_name
=
first_name
,
last_name
=
last_name
)
return
players
Remember that Table 4-1 defined the endpoints that we planned to create as a combination of HTTP verb and URL. With FastAPI these endpoints (also called routes) are defined with the decorators above each function.
The following explains how the HTTP verb and URL are specified in the decorator:
-
HTTP verb: All of these endpoints use the
GET
verb, which is defined by the@app.get()
decorator function. -
URL: The first parameter of the
get()
function is the relative URL. For this first endpoint, the URL is /v0/players/.
The second parameter of the decorator is response_model=list[schemas.Player])
. This informs FastAPI that the data returned from this endpoint will be a list of Pydantic Player
objects, as defined in the schemas.py file. This information will be included in the OpenAPI specification that FastAPI automatically creates for this API. Consumers can count on the returned data being valid according to this definition.
Let’s look at the function signature that you decorated:
def
read_players
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
=
None
,
first_name
:
str
=
None
,
last_name
:
str
=
None
,
db
:
Session
=
Depends
(
get_db
)):
Several things are going on in this function. Starting at the end, the db
object is a session that is created by the get_db()
function defined at the top of this file. By wrapping the function in Depends()
, FastAPI handles the call for and gives the Session
to your function.
The next two parameters are optional integers with a default value: skip: int = 0, limit: int = 100, last_name. These are followed by two optional string parameters that default to +None
. These are all named parameters that have a defined datatype and a default value. FastAPI will automatically include these parameters as query parameters in the API definition. Query parameters are included in the URL path with a question mark in front, and an ampersand between.
For instance to call this query method, the API consumer could use this request:
-
HTTP verb: GET
-
URL: {base URL}/v0/players/?skip=10&limit=50&minimum_last_changed_date=2024-04-01
Or the API consumer could use this request:
-
HTTP verb: GET
-
URL: {base URL}/v0/players/?first_name=Bryce&last_name=Young
Within the body of the read_players()
function, FastAPI is calling the get_players()
function that you defined in crud.py. It is performing a database query. The players
object receives the result of that function call. FastAPI validates that this object matches the definition list[schemas.Player]
. If it does, FastAPI uses Pydantic to serialize the Python objects into a text JSON string and sends the response to the consumer.
The next endpoint adds two additional FastAPI features:
@app
.
get
(
"/v0/players/
{player_id}
"
,
response_model
=
schemas
.
Player
)
def
read_player
(
player_id
:
int
,
db
:
Session
=
Depends
(
get_db
)):
player
=
crud
.
get_player
(
db
,
player_id
=
player_id
)
if
player
is
None
:
raise
HTTPException
(
status_code
=
404
,
detail
=
"Player not found"
)
return
player
First, the URL path includes {player_id}
. This is a path parameter, which is an API request parameter that is included in the URL path instead of separated by question marks and ampersands like the query parameters. Here is an example of how the API consumer might call this endpoint:
-
HTTP verb: GET
-
URL: {base URL}/v0/players/12345?skip=10&limit=50
The function checks to see if any records were returned from the helper function, and if not it raises an HTTPException. This is a standard method that web applications use to communicate status. It is good RESTful API design to use the standard HTTP status codes to communicate with consumers. This makes the operation more predictable and reliable. This endpoint returns an HTTP status code of 404, which is the not found code. It adds the additional message that the item not found was the player being searched for.
The next four endpoints do not use any new features. But together they complete all of the user stories that we have included for your first API:
@app
.
get
(
"/v0/performances/"
,
response_model
=
list
[
schemas
.
Performance
])
def
read_performances
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
=
None
,
db
:
Session
=
Depends
(
get_db
)):
performances
=
crud
.
get_performances
(
db
,
skip
=
skip
,
limit
=
limit
,
min_last_changed_date
=
minimum_last_changed_date
)
return
performances
@app
.
get
(
"/v0/leagues/
{league_id}
"
,
response_model
=
schemas
.
League
)
def
read_league
(
league_id
:
int
,
db
:
Session
=
Depends
(
get_db
)):
league
=
crud
.
get_league
(
db
,
league_id
=
league_id
)
if
league
is
None
:
raise
HTTPException
(
status_code
=
404
,
detail
=
"League not found"
)
return
league
@app
.
get
(
"/v0/leagues/"
,
response_model
=
list
[
schemas
.
League
])
def
read_leagues
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
=
None
,
league_name
:
str
=
None
,
db
:
Session
=
Depends
(
get_db
)):
leagues
=
crud
.
get_leagues
(
db
,
skip
=
skip
,
limit
=
limit
,
min_last_changed_date
=
minimum_last_changed_date
,
league_name
=
league_name
)
return
leagues
@app
.
get
(
"/v0/teams/"
,
response_model
=
list
[
schemas
.
Team
])
def
read_teams
(
skip
:
int
=
0
,
limit
:
int
=
100
,
minimum_last_changed_date
:
date
=
None
,
team_name
:
str
=
None
,
db
:
Session
=
Depends
(
get_db
)):
teams
=
crud
.
get_teams
(
db
,
skip
=
skip
,
limit
=
limit
,
min_last_changed_date
=
minimum_last_changed_date
,
team_name
=
team_name
)
return
teams
The final endpoint provides counts of leagues, teams, and players:
@app
.
get
(
"/v0/counts/"
,
response_model
=
schemas
.
Counts
)
def
get_count
(
db
:
Session
=
Depends
(
get_db
)):
counts
=
schemas
.
Counts
(
league_count
=
crud
.
get_league_count
(
db
),
team_count
=
crud
.
get_team_count
(
db
),
player_count
=
crud
.
get_player_count
(
db
))
return
counts
It is worth noting that in addition to the basic options of FastAPI and Pydantic that you are using, there are many additional validations and other features available. As you can see, these libraries accomplish a lot with only a few lines of code from you.
Testing Your API
You will use Pytest to test your main.py
. As with the crud.py
file in the previous chapter, you will be testing that the correct number of records are returned by each API endpoint. The counts of records can be verified by the SQL queries in Chapter 3’s “Loading Your Data” section.
To implement the tests for your API, create a file with the following contents named test_main.py:
from
fastapi.testclient
import
TestClient
from
main
import
app
client
=
TestClient
(
app
)
# test the health check endpoint
def
test_read_main
():
response
=
client
.
get
(
"/"
)
assert
response
.
status_code
==
200
assert
response
.
json
()
==
{
"message"
:
"API health check successful"
}
# test /v0/players/
def
test_read_players
():
response
=
client
.
get
(
"/v0/players/?skip=0&limit=10000"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
1018
def
test_read_players_by_name
():
response
=
client
.
get
(
"/v0/players/?first_name=Bryce&last_name=Young"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
1
assert
response
.
json
()[
0
]
.
get
(
"player_id"
)
==
2009
# test /v0/players/{player_id}/
def
test_read_players_with_id
():
response
=
client
.
get
(
"/v0/players/1001/"
)
assert
response
.
status_code
==
200
assert
response
.
json
()
.
get
(
"player_id"
)
==
1001
# test /v0/performances/
def
test_read_performances
():
response
=
client
.
get
(
"/v0/performances/?skip=0&limit=20000"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
17306
# test /v0/performances/ with changed date
def
test_read_performances_by_date
():
response
=
client
.
get
(
"/v0/performances/?skip=0&limit=20000&minimum_last_changed_date=2024-04-01"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
2711
# test /v0/leagues/{league_id}/
def
test_read_leagues_with_id
():
response
=
client
.
get
(
"/v0/leagues/5002/"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
()[
"teams"
])
==
8
# test /v0/leagues/
def
test_read_leagues
():
response
=
client
.
get
(
"/v0/leagues/?skip=0&limit=500"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
5
# test /v0/teams/
def
test_read_teams
():
response
=
client
.
get
(
"/v0/teams/?skip=0&limit=500"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
20
# test /v0/teams/
def
test_read_teams_for_one_league
():
response
=
client
.
get
(
"/v0/teams/?skip=0&limit=500&league_id=5001"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
12
# test the count functions
def
test_counts
():
response
=
client
.
get
(
"/v0/counts/"
)
response_data
=
response
.
json
()
assert
response
.
status_code
==
200
assert
response_data
[
"league_count"
]
==
5
assert
response_data
[
"team_count"
]
==
20
assert
response_data
[
"player_count"
]
==
1018
To recap the previous chapter, any file that contains tests will have a file name beginning with test_ or ending with test. Inside the test file, Pytest will execute any function name beginning with test_. These functions will end with an assert
statement. If it’s true, the test passes, if it’s false the test fails.
The file begins with import statements and the creation the TestClient
:
from
fastapi
.
testclient
import
TestClient
from
main
import
app
client
=
TestClient
(
app
)
The
TestClient
is a special class that allows the FastAPI program to be tested without running it on a web server.The
main.app
is the main FastAPI program you created in main.py.This statement creates a
TestClient
that will test your application.
Take a look at a few of the test functions:
#test the health check endpoint
def
test_read_main
():
response
=
client
.
get
(
"/"
)
assert
response
.
status_code
==
200
assert
response
.
json
()
==
{
"message"
:
"API health check successful"
}
This function uses the TestClient
to simulate an API call to the root path. Then it checks the HTTP status code for a value of 200, which means a successful request. Next, it looks at the JSON value returned by the API and checks that it matches the JSON value provided.
The next test function adds more functionality:
#test /v0/players/
def
test_read_players
():
response
=
client
.
get
(
"/v0/players/?skip=0&limit=10000"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
1018
Notice that the URL passed in the get()
statement uses the skip
and limit+parameters. The second +assert
statement checks the length of the list players returned by the API to make sure it is exactly 1018.
Another test function tests the search of players by name. Although the database does not enforce uniqueness on player names, duplicate player names are rare, and names are commonly used to identify players.
This search without a key supports the design recommended in Chapter 1 for AI:
def
test_read_players_by_name
():
response
=
client
.
get
(
"/v0/players/?first_name=Bryce&last_name=Young"
)
assert
response
.
status_code
==
200
assert
len
(
response
.
json
())
==
1
assert
response
.
json
()[
0
]
.
get
(
"player_id"
)
==
2009
This adds two assert
statements: one to make sure only one record was returned from this query. (After all, there is only one Bryce Young) and another to make sure the player_id
is correct.
The complete file contains 11 tests in all. To execute the tests, enter the following command:
.../chapter4
(
main)
$
pytest
test_main.py
==================
test
session
starts
===========================
platform
linux
--
Python
3
.10.14,pytest-8.1.2,
pluggy-1.4.0
rootdir:
/workspaces/portfolio-project/chapter4
plugins:
anyio-3.4.4.0
collected
11
items
test_main.py
[
100
%]
===================
11
passed
in
1
.01s
============================
You have verified that your FastAPI program works with Pytest. Now it’s time to try it with a web server.
Launching Your API
This is the moment you have been waiting for: it’s time to run your API. You are going to use the Uvicorn web server to execute your program, passing the name of the Python file without the file extension and the FastAPI app object name.
Enter the following command from the command line:
.../chapter4
(
main)
$
fastapi
run
main.py
You will see the application startup occur as show in Figure 4-2.
In Codespaces, you will also see a dialog stating “Your application running on port 8000 is available” as shown in Figure 4-3.
Click “Open in Browser” to open a browser tab outside of your codespaces. This browser will show a base URL ending in “app.github.dev” that contains the response from your API running on Codespaces. You should see the health check message in your web browser:
{"message":"API health check successful"}
This confirms your API is running, which is a great start.
The next test is to call an endpoint that retrieves data. Give that a try by copying and pasting the following onto the end of the base URL in your browser: /v0/performances/?skip=0&limit=1. For example, the full URL might be https://happy-pine-tree-1234-8000.app.github.dev/v0/performances/?skip=0&limit=1.
If everything is working correctly, you should see the following data in your browser:
[{
"performance_id"
:
2501
,
"player_id"
:
1001
,
"week_number"
:
"202301"
,
"fantasy_points"
:
20.0
,
"last_changed_date"
:
"2024-03-01"
}]
Tip
This chapter covered a lot, so it’s possible that an error occurred or you are not getting a successful result. Don’t worry, this happens to all of us. Here are a few suggestions for how to troubleshoot any problems you are running into:
-
Run the pip3 install -r requirements.txt command again to make sure you have all the updated software.
-
Take a minute to verify the path in the URL bar of your browser. Minor things matter such as slashes and question marks.
-
Look at the command line to see any errors that are being thrown by FastAPI.
-
To verify your environment with FastAPI and Uvicorn, try creating a very simple API, such as one from the Official FastAPI Tutorial.
If this first API endpoint is working for you, try out some more of the URLs from Table 4-1 in your browser to verify that you have completed all of your user stories. Congratulations, you are an API developer!
Additional Resources
To explore FastAPI beyond this book, the official FastAPI tutorial and FastAPI reference documentation are both very useful.
To learn the ins and outs of building a project with FastAPI, I recommend FastAPI: Modern Python Web Development by Bill Lubanovic (O’Reilly, 2023).
For a growing list of practical tips from an official FastAPI Export, check out Marcelo Trylesinski’s Fast API Tips.
The official Pydantic 2.4 documentation provides information for the specific version of Pydantic used in this chapter.
The official Uvicorn documentation has much more information about the capabilities of this software.
Summary
In this chapter, you completed the API functionality for the SWC Fantasy Football API. You accomplished the following:
-
You installed FastAPI, SQLAlchemy, Pydantic, and Uvicorn, along with several supporting libraries.
-
You created SQLAlchemy models and CRUD helper functions to process the data using Python.
-
You defined Pydantic schemas to represent the data that your API consumers wanted to receive.
-
You created a FastAPI program to process consumer requests and return data responses, tying everything together.
-
You tested the API with Pytest and then ran it successfully on the web server.
In Chapter 5, you will look at how to document your API using FastAPI’s built-in capabilities and then publish a developer portal.
Get Hands-On APIs for AI and Data Science now with the O’Reilly learning platform.
O’Reilly members experience books, live events, courses curated by job role, and more from O’Reilly and nearly 200 top publishers.