It's been a while since I've had the opportunity to build something simple, interesting and modern. Towards the backend of 2024 I stumbled across FastAPI and got excited, whilst I've built internal APIs at work before, I hadn't yet created anything public facing.
Hello FastAPI!
FastAPI is a modern, powerful framework for building APIs with Python and it seemed perfect for what I wanted to build, an API for basic football player info. I initially dubbed it "Jugador FC" before settling for "Player FC API".
Configuring Environment.
Before you begin, make sure you have the following requirements in place:
AWS CDK
Docker
Python 3.12.7
Creating the Project
Create a directory on your machine. Name it player_fc_fastapi_app, within this directory create the following subdirectories:
app
Contains all the FastAPI code
dynamo_db_local
Contains a python script to create a local version of an Amazon DynamoDB Table
iac
Contains your stack files to create resources in AWS
I have made it easier by providing the commands that you can run to save time below:
mkdir player_fc_fastapi_app && cd player_fc_fastapi_app | |
mkdir app dynamo_db_local iac |
The project directory structure should now look like below:
├── player_fc_fastapi_app | |
│ ├── app | |
│ └── dynamo_db_local | |
│ └── iac | |
Setting up the Python environment
After creating the directory structure, create a text file called requirements.txt
and insert the following lines in it:
fastapi | |
uvicorn | |
mangum | |
pydantic | |
boto3 | |
pytest | |
httpx |
Once you have created the requirements.txt
file, create a virtual environment and install the dependencies:
python3 -m venv .venv | |
# macOS | |
source .venv/bin/activate | |
# Windows | |
.\.venv\Scripts\activate | |
pip install -r requirements.txt |
Setting up Amazon DynamoDB Local
Let's begin with setting up a local instance of DynamoDB, this will require Docker to be installed and running.
docker run --rm -p 7001:8000 amazon/dynamodb-local |
This will take a few seconds for the image to be pulled and starting a container, once done we can navigate towards the dynamo_db_local directory and create a create_ddb_table.py
file, populate the file with the below code:
import boto3 | |
from botocore.exceptions import ClientError | |
def main(): | |
# Create DynamoDB resource | |
dynamodb = boto3.resource( | |
"dynamodb", | |
endpoint_url="http://localhost:7001", | |
region_name="af-south", | |
aws_access_key_id="myid", | |
aws_secret_access_key="myaccesskey", | |
) | |
create_dynamodb_table(dynamodb) | |
def create_dynamodb_table(dynamodb): | |
# Check if DynamoDB Table exists | |
table_name = "Players" | |
existing_tables = dynamodb.meta.client.list_tables()["TableNames"] | |
if table_name not in existing_tables: | |
try: | |
# Create DynamoDB Table | |
table = dynamodb.create_table( | |
TableName=table_name, | |
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}], | |
AttributeDefinitions=[ | |
{"AttributeName": "id", "AttributeType": "S"}], | |
BillingMode='PAY_PER_REQUEST' | |
) | |
print(f"Successfully created table: {table.table_name}") | |
except ClientError as e: | |
print(e) | |
else: | |
print(f"Table '{table_name}' already exists.") | |
if __name__ == "__main__": | |
main() |
With this code, you can create a table in the local DynamoDB instance. Run the code snippet.
FastAPI Development
Now that we have a local instance of DynamoDB up and running, let's begin creating our app, navigate towards the app directory and create two files, main.py
and requirements.txt
.
Populate the requirements.txt
with the below:
mangum | |
fastapi | |
pydantic |
Create the below subdirectories :
models
Pydantic Player models
routers
Contains routes
├── player_fc_fastapi_app | |
│ ├── app | |
│ │ ├── models | |
| | | ├── __init__.py | |
| | | └── players.py | |
│ │ └── routers | |
| | | ├── __init__.py | |
| | | └── players.py | |
| | ├── main.py | |
│ │ └── requirements.txt | |
│ ├── dynamo_db_local | |
│ ├── iac | |
│ └── requirements.txt |
Let's create a couple of models using Pydantic, we will use the Player
and UpdatePlayer
models to define the data structure of player info we can add or modify.
Within the models subdirectory, create an empty __init__.py
file and a file named players.py
and fill with the below code:
from pydantic import BaseModel | |
from datetime import date | |
from typing import Optional | |
class Player(BaseModel): | |
name: str | |
country: str | |
date_of_birth: date | |
team: str | |
position: str | |
club_number: int | |
national_team_number: int | |
class UpdatePlayer(BaseModel): | |
team: Optional[str] = None | |
position: Optional[str] = None | |
club_number: Optional[int] = None | |
national_team_number: Optional[int] = None |
Within the routers subdirectory, create an empty __init__.py
file and a file named players.py
and fill with the below code:
import botocore.exceptions | |
import boto3 | |
import botocore | |
import uuid | |
import logging | |
from fastapi import FastAPI, HTTPException, status | |
from models.players import Player, UpdatePlayer | |
from fastapi import APIRouter | |
router = APIRouter() | |
def get_dynamodb_table(local_development: bool = True): | |
"""Retrieve DynamoDB Table connection based on environment""" | |
table_name = "Players" | |
if local_development: | |
return boto3.resource("dynamodb", | |
endpoint_url="http://localhost:7001", | |
region_name="af-south", | |
aws_access_key_id="myid", | |
aws_secret_access_key="myaccesskey").Table(table_name) | |
else: | |
return boto3.resource("dynamodb").Table(table_name) | |
@router.post("/players") | |
async def create_player(player: Player) -> dict: | |
"""Create player in DynamoDB Table""" | |
player_id = uuid.uuid5( | |
uuid.NAMESPACE_DNS, | |
f"{player.name}-{player.country}" | |
) | |
item = { | |
"id": str(player_id), | |
"name": player.name, | |
"country": player.country, | |
"date_of_birth": player.date_of_birth.isoformat(), | |
"team": player.team, | |
"position": player.position, | |
"club_number": player.club_number, | |
"national_team_number": player.national_team_number | |
} | |
# Add player to DynamoDB Table | |
try: | |
table = get_dynamodb_table() | |
table.put_item(Item=item) | |
return {"player_id": item["id"], "player_name": item["name"]} | |
except botocore.exceptions.ClientError as e: | |
logging.exception(f"An error occurred: {e}") | |
raise HTTPException(status_code=500, detail="An error occurred creating player.") | |
@router.get("/players") | |
async def get_all_players(): | |
"""Retrieve all players from DynamoDB Table""" | |
table = get_dynamodb_table() | |
try: | |
response = table.scan() | |
items = response["Items"] | |
if len(items) == 0: | |
return {"message": "No players found"} | |
else: | |
return {"count": len(items), "players": items} | |
except botocore.exceptions.ClientError as e: | |
logging.exception(f"An error occurred: {e}") | |
raise HTTPException(status_code=500, detail="An error occurred retrieving all players.") | |
@router.get("/players/{id}") | |
async def get_player(id: str): | |
"""Retrieve player data by id from DynamoDB Table""" | |
table = get_dynamodb_table() | |
response = table.get_item(Key={"id": id}) | |
item = response.get("Item") | |
if not item: | |
raise HTTPException(status_code=404, detail=f"Player '{id}' not found") | |
return item | |
@router.patch("/players/{id}") | |
async def update_player(id: str, player: UpdatePlayer): | |
"""Update player details in DynamoDB Table""" | |
table = get_dynamodb_table() | |
# Check if id exists before performing patch operation | |
response = table.get_item(Key={"id": id}) | |
item = response.get("Item") | |
if not item: | |
raise HTTPException(status_code=404, detail=f"Player '{ | |
id}' does not exist") | |
update_fields = { | |
key: value for key, value in player | |
if value is not None | |
} | |
# Dynamically build 'UpdateExpression' | |
update_expression = "set " + \ | |
", ".join(f"#{key} = :{key}" for key in update_fields.keys()) | |
# Dynamically build 'ExpressionAttributeNames' | |
expression_attribute_names = {f"#{key}": str( | |
key) for key in update_fields.keys()} | |
# Dynamically build 'ExpressionAttributeValues' | |
expression_attribute_values = { | |
f":{key}": value for key, value in update_fields.items()} | |
try: | |
response = table.update_item( | |
Key={"id": id}, | |
UpdateExpression=update_expression, | |
ExpressionAttributeNames=expression_attribute_names, | |
ExpressionAttributeValues=expression_attribute_values, | |
ReturnValues="UPDATED_NEW", | |
) | |
return {"id": id, "attributes": response["Attributes"]} | |
except botocore.exceptions.ClientError as e: | |
logging.exception(f"An error occurred: {e}") | |
@router.delete("/players/{id}") | |
async def delete_player(id: str): | |
"""Delete player from DynamoDB Table""" | |
table = get_dynamodb_table() | |
try: | |
response = table.get_item(Key={"id": id}) | |
if "Item" not in response: | |
raise HTTPException( | |
status_code=404, detail=f"Player '{id}' not found") | |
table.delete_item(Key={ | |
"id": id | |
}) | |
return {"message": f"'{id}' successfully deleted."} | |
except HTTPException: | |
raise | |
except Exception as e: | |
raise HTTPException(status_code=500, detail="An error occurred deleting player.") | |
Creating an empty
__init__.py
file turns a folder into a Python package.
Create a file named main.py
within the app subdirectory and start populating it with the below code:
from fastapi import FastAPI | |
from mangum import Mangum | |
from routers.players import router as players_router | |
import logging | |
logger = logging.getLogger() | |
logger.setLevel("INFO") | |
app = FastAPI() | |
app.include_router(players_router) | |
handler = Mangum(app) | |
@app.get("/") | |
def root(): | |
return {"message": "Welcome to the Player FC API!", | |
"description": "More than just a Game", | |
} |
Test Drive
Time for a quick test drive, ensure you are in the app directory and run the below command to start Uvicorn:
uvicorn main:app --reload |
Now that our app is up and running, navigate to http://127.0.0.1:8000/docs/
You will see the automatic interactive API documentation with 6 endpoints available:
Let's try adding a player. Select the POST /players endpoint, select the Try It out button and use the below payload to add the world's best player, "Vinícius Júnior":
{ | |
"name": "Vinícius Júnior", | |
"country": "Brazil", | |
"date_of_birth": "2000-07-12", | |
"team": "Real Madrid", | |
"position": "Forward", | |
"club_number": 7, | |
"national_team_number": 7 | |
} |
Here's what each API operation looks like in action.
Adding a New Player:
Retrieving All Players:
Updating Player Information:
Getting Single Player Details:
Removing a Player:
Deployment using AWS CDK v2
Now that we are comfortable with running and testing our app locally, it's time to deploy our app on AWS. We will use the AWS CDK v2.
Navigate into the iac directory, run the below command to initialize a cdk project:
cdk init --language python |
Modify the requirements.txt file found in the subdirectory, add the below line:
aws-cdk.aws-lambda-python-alpha |
Let's define a DynamoDB Table, Lambda function and a Lambda function url. In the current iac directory, there is another subdirectory that you need to navigate towards (iac). Open the iac_stack.py
file and replace the contents of the CDK stack with the code below:
from aws_cdk import ( | |
Stack, | |
Duration, | |
RemovalPolicy, | |
aws_dynamodb as dynamodb, | |
aws_lambda as _lambda, | |
aws_lambda_python_alpha as _alambda, | |
CfnOutput | |
) | |
from constructs import Construct | |
class IacStack(Stack): | |
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None: | |
super().__init__(scope, construct_id, **kwargs) | |
# Create DynamoDB Table | |
table = dynamodb.TableV2(self, "Table", | |
table_name="Players", | |
partition_key=dynamodb.Attribute( | |
name="id", type=dynamodb.AttributeType.STRING), | |
removal_policy=RemovalPolicy.DESTROY, | |
) | |
# Create Lambda function for API endpoint | |
api = _alambda.PythonFunction( | |
self, | |
"API", | |
entry="../app", | |
function_name="player_fc_api", | |
runtime=_lambda.Runtime.PYTHON_3_12, | |
index="main.py", | |
handler="handler", | |
timeout=Duration.seconds(60) | |
) | |
# Create Lambda Function URL | |
functionUrl = api.add_function_url( | |
auth_type=_lambda.FunctionUrlAuthType.NONE, | |
cors=_lambda.FunctionUrlCorsOptions( | |
allowed_origins=["*"], | |
allowed_methods=[_lambda.HttpMethod.ALL], | |
allowed_headers=["*"] | |
) | |
) | |
# Print the Function URL | |
CfnOutput(self, "FunctionUrl", value=f"{functionUrl.url}docs") | |
# Permissions for Function to access DynamoDB | |
table.grant_read_write_data(api) |
We have one final step before we initiate the deploy, set the flag for local_development: bool
to False in the players.py
file in the app/routers directory.
def get_dynamodb_table(local_development: bool = False): | |
"""Retrieve DynamoDB Table connection based on environment""" | |
table_name = "Players" | |
if local_development: | |
return boto3.resource("dynamodb", | |
endpoint_url="http://localhost:7001", | |
region_name="af-south", | |
aws_access_key_id="myid", | |
aws_secret_access_key="myaccesskey").Table(table_name) | |
else: | |
return boto3.resource("dynamodb").Table(table_name) |
Activate the virtual environment within the iac directory and install the dependencies with the below commands:
# macOS | |
source .venv/bin/activate | |
# Windows | |
.\.venv\Scripts\activate | |
pip install -r requirements.txt |
Deploy the app with the cdk deploy
command.
Once the deployment is complete, you'll see a function URL in the terminal output, this is your API endpoint on AWS.
Test all endpoints using the function URL like we did during the local test drive. Once you have added a player it's time to verify if our player data has persisted or vanished into the ether.
To verify everything's working:
- Head over to the AWS Management Console
- Navigate to DynamoDB
- Find the Players Table
- Select Explore table items
You should see your player data in the cloud:
💡 Important: Don't forget to clean up resources! When no longer needed, you can run the cdk destroy
command to delete all AWS resources that were created.
That wraps up our journey from local FastAPI development to serverless deployment on AWS.
Top comments (0)