Building Your First API with FastAPI

Introduction

Welcome to this beginner-friendly guide on embarking upon the realms of web development by building a simple yet robust API with Python and FastAPI. In this tutorial, we will delve into the process of setting up a FastAPI project, creating all the necessary endpoints for a TODO application, including GET, POST, PUT, and DELETE requests.

We’ve chosen SQLite as our database to simplify the persistence layer, allowing you to focus more on understanding the FastAPI framework and less on database configurations. SQLite is a self-contained, serverless, and zero-configuration database engine, which makes it a great choice for beginners and for the purpose of this tutorial.

…
By the end of this tutorial, you’ll have a working web API deployed locally on your machine. The complete code for this tutorial is available on GitHub.

Why 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, efficient, and scalable. FastAPI comes with built-in support for documentation generation and automated testing, making it an excellent choice for building robust web applications.

Setting Up Your Environment

Before we dive into building our TODO application, let’s make sure we have all the necessary tools and libraries installed. First, make sure you have Python 3.7 or higher installed on your machine. You can check your Python version by running the following command in your terminal:

python --version

If you don’t have Python installed, you can download and install it from the official Python website.

Once you have Python installed, you can create a virtual environment for your project. Open your terminal and navigate to the directory where you want to create your project. Then run the following commands:

python -m venv myenv
source myenv/bin/activate

This will create a new virtual environment and activate it. Now, you can install FastAPI, Uvicorn, SQLAlchemy, and other required libraries by running the following command:

pip install fastapi uvicorn sqlalchemy

Configuring the Database

We will use SQLite for this tutorial. Let’s start by setting up our database and ORM. Create a new file database.py and add the following code:

# database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

Base = declarative_base()

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

def init_db():
    Base.metadata.create_all(bind=engine)

DATABASE_URL = "sqlite:///./test.db"

this line sets up a constant named DATABASE_URL that holds the connection string to the database. In this case, it’s connecting to a SQLite database named test.db located in the current directory.

Base = declarative_base():

This line creates a base class for declarative class definitions. The declarative_base() function returns a new base class from which all mapped classes should inherit. By using the declarative system, you can create classes that include directives to describe the actual database table they will be mapped to.

engine = create_engine(DATABASE_URL):

This line creates a new instance of an engine, which is the source of connectivity to the database. The create_engine() function takes the database URL, in this case DATABASE_URL, and returns an instance of Engine. This engine will be used to interact with the database.

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

This line creates a custom sessionmaker, which is a factory for producing session objects, which are the SQLAlchemy equivalent of a “workspace” for our operations.
The autocommit argument is set to False to ensure that the session does not commit automatically, allowing you to have better control over when transactions are committed to the database.
The autoflush argument is set to False to ensure the session does not flush automatically, which means that it will not synchronize the session’s state with the database unless explicitly told to do so.
The bind argument binds the sessionmaker to our engine, so that the sessions it creates are connected to our database.

def init_db()::

This line defines a function named init_db which will be used to initialize the database.

Base.metadata.create_all(bind=engine):

Inside the init_db function, this line creates all of the tables defined on the Base class and its subclasses, which includes any Table objects that might be associated with the Base and its subclasses.
The bind argument specifies the engine that will be used to talk to the database, so this line is essentially telling SQLAlchemy to create all of the necessary database tables using the specified engine.

This setup is a simplified way of establishing a connection to a database, defining a way to create new sessions (which represent transactions), and initializing the database with the required tables.

Defining Database Models

In order to interact with the database, we need to define our data models. We’ll create a new file named models.py where we’ll define a SQLAlchemy model for our TODO items. This way, we keep our project organized and separate the database schema definition from our main application logic.

Create a new file models.py and add the following code:

# models.py
from sqlalchemy import Column, Integer, String, Boolean
from database import Base

class TodoInDB(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, index=True)
    task = Column(String, index=True)
    completed = Column(Boolean, default=False)

Now, TodoInDB is our database model representing a TODO item with three fields: id, task, and completed. The id field is our primary key, task is a string representing the task description, and completed is a boolean indicating whether the task has been completed or not.

Creating the TODO Application

Now that we have our environment and database set up, let’s start building our TODO application. We will create a new Python file called main.py and import the necessary modules:

# main.py
from fastapi import FastAPI, HTTPException, status, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from models import TodoInDB
from database import SessionLocal, init_db

app = FastAPI()

@app.on_event("startup")
async def startup():
    init_db()

class Todo(BaseModel):
    task: str
    completed: bool

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def to_dict(obj):
    return {column.name: getattr(obj, column.name) for column in obj.__table__.columns}

This will be the structure of our TODO objects. We have defined three fields: `task`, and `completed`.

app = FastAPI():

This line creates an instance of the FastAPI class. This instance will serve as the web application, and will provide all of the functionality for defining endpoints, handling requests, and returning responses.

@app.on_event("startup"):

This line is a decorator that registers a function to be run when the FastAPI application starts up. It’s part of FastAPI’s event handling system.

async def startup()::

This line defines a coroutine named startup. This coroutine will be executed when the FastAPI application starts up, thanks to the @app.on_event("startup") decorator above it.

init_db():

Inside the startup coroutine, this line calls the init_db function defined earlier to initialize the database.

class Todo(BaseModel)::

This line defines a new class named Todo, which inherits from BaseModel (a class from the Pydantic library). This class will serve as the schema for Todo objects throughout the application.

task: str and completed: bool:

These lines define the fields of the Todo schema: a task field of type str, and a completed field of type bool.

def get_db()::

This line defines a function named get_db, which will be used to get a database session.

db = SessionLocal():

Inside get_db, this line creates a new database session using the SessionLocal session factory defined earlier.

try: and finally::

These lines establish a try/finally block, which ensures that the code in the finally block will be executed no matter what, even if an exception occurs.

yield db:

Inside the try block, this line yields the database session to the caller. This is a way of providing the database session to other parts of the code while ensuring that the finally block will be executed to clean up the session afterwards.

db.close():

Inside the finally block, this line closes the database session. This is important to prevent resource leaks by ensuring that every session is closed when it’s no longer needed.

def to_dict(obj)::

This line defines a function named to_dict, which will be used to convert SQLAlchemy ORM objects to dictionaries.

return {column.name: getattr(obj, column.name) for column in obj.__table__.columns}:

Inside to_dict, this line returns a dictionary that contains all of the data from the given ORM object. It does this by iterating over all of the columns of the object’s table, getting the value of each column from the object, and putting those values into a dictionary.

GET Request – Retrieving All TODOs

Now, let’s create our first endpoint for retrieving all TODOs. Add the following code to your `main.py` file:

@app.get("/todos", response_model=list[dict])
def read_todos(db: Session = Depends(get_db)):
    todos = db.query(TodoInDB).all()
    return [to_dict(todo) for todo in todos]

This endpoint will return all the TODOs stored in our application. We will define the `todos` variable later in the code.

@app.get("/todos", response_model=list[dict]):

This is a decorator provided by FastAPI to declare a new HTTP GET endpoint at the path /todos. The response_model parameter is a Pydantic model (or in this case, a Python type hint) that describes the shape of the response data. Here, list[dict] indicates that the endpoint will return a list of dictionaries.

def read_todos(db: Session = Depends(get_db))::

This line defines a new function read_todos which will handle requests to the /todos endpoint. The db: Session = Depends(get_db) argument instructs FastAPI to call the get_db function (described in the previous code snippet) to obtain a database session, and to pass that session to read_todos as the db argument.

todos = db.query(TodoInDB).all():

Inside read_todos, this line creates a new SQLAlchemy query object that will query the TodoInDB model. The .all() method is then called on this query object to execute the query and return all rows from the todos table as a list of TodoInDB objects.

return [to_dict(todo) for todo in todos]:

This line is a list comprehension that iterates over each TodoInDB object in the todos list, calls the to_dict function (described in the previous code snippet) to convert each TodoInDB object to a dictionary, and then returns the list of dictionaries. This list of dictionaries will be serialized to JSON and sent back to the client as the HTTP response.

POST Request – Creating a New TODO

To create a new TODO, we need to define another endpoint. Add the following code to your `main.py` file:

@app.post("/todos", response_model=dict, status_code=status.HTTP_201_CREATED)
def create_todo(todo: Todo, db: Session = Depends(get_db)):
    db_todo = TodoInDB(**todo.model_dump())
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return to_dict(db_todo)

This endpoint expects a JSON object representing a TODO and adds it to our list of TODOs. We will define the `todos` list later in the code.

@app.post("/todos", response_model=dict, status_code=status.HTTP_201_CREATED):

This line is a decorator provided by FastAPI to declare a new HTTP POST endpoint at the path /todos. The response_model parameter is a Pydantic model (or in this case, a Python type hint) that describes the shape of the response data. Here, dict indicates that the endpoint will return a dictionary. The status_code parameter sets the HTTP status code to 201 (Created) for successful responses from this endpoint.

def create_todo(todo: Todo, db: Session = Depends(get_db))::

This line defines a new function create_todo to handle requests to the /todos endpoint. It accepts two arguments: todo of type Todo (which is a Pydantic model you defined earlier), and db of type Session. FastAPI will call the get_db function to obtain a database session and pass it as the db argument, as specified by the Depends keyword.

db_todo = TodoInDB(**todo.model_dump()):

This line creates a new TodoInDB object from the data in the todo argument. It uses the ** (unpacking) operator to pass the fields of the todo object as keyword arguments to the TodoInDB constructor, and model_dump() to convert the todo object to a dictionary.

db.add(db_todo):

This line adds the new TodoInDB object to the current database session, marking it as pending to be inserted into the database.

db.commit():

This line commits the current database session, flushing all pending changes to the database. This will insert the new TodoInDB object into the todos table.

db.refresh(db_todo):

This line refreshes the db_todo object with the current state in the database, ensuring it reflects any changes that occurred during the commit (such as the assignment of a new id value by the database).

return to_dict(db_todo):

This line calls the to_dict function (explained in a previous snippet) to convert the db_todo object to a dictionary, and returns this dictionary. This dictionary will be serialized to JSON and sent back to the client as the HTTP response.

PUT Request – Updating a TODO

Next, let’s create an endpoint for updating a TODO. Add the following code to your `main.py` file:

@app.put("/todos/{todo_id}", response_model=dict)
def update_todo(todo_id: int, todo: Todo, db: Session = Depends(get_db)):
    db_todo = db.query(TodoInDB).filter(TodoInDB.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    for key, value in todo.model_dump().items():
        setattr(db_todo, key, value)
    db.commit()
    db.refresh(db_todo)
    return to_dict(db_todo)

This endpoint expects an `id` parameter in the URL and a JSON object representing the updated TODO. It updates the TODO with the specified `id` if it exists in our list of TODOs.

@app.put("/todos/{todo_id}", response_model=dict):

This line is a FastAPI decorator that defines a new HTTP PUT endpoint at the path /todos/{todo_id}. The response_model parameter indicates that the endpoint will return a dictionary. The {todo_id} in the path is a path parameter that will be passed to the update_todo function.

def update_todo(todo_id: int, todo: Todo, db: Session = Depends(get_db))::

This line declares the update_todo function which will handle requests to the /todos/{todo_id} endpoint. It accepts three arguments: todo_id of type int, todo of type Todo, and db of type Session. FastAPI will call the get_db function to obtain a database session and pass it as the db argument, as specified by the Depends keyword.

db_todo = db.query(TodoInDB).filter(TodoInDB.id == todo_id).first():

This line queries the database for a TodoInDB object with the specified todo_id. It uses SQLAlchemy’s query interface to construct and execute the query.

if db_todo is None: and raise HTTPException(status_code=404, detail="Todo not found"):

These lines check if the query returned a TodoInDB object. If not, it raises an HTTPException with a status code of 404 (Not Found) and a detail message of “Todo not found”.

for key, value in todo.model_dump().items(): and setattr(db_todo, key, value):

These lines iterate over the fields of the todo object, updating the corresponding fields of the db_todo object with the new values. It uses the model_dump() method to convert the todo object to a dictionary, and the setattr function to update the db_todo object.

db.commit():

This line commits the current database session, saving the changes to the db_todo object to the database.

db.refresh(db_todo):

This line refreshes the db_todo object with the current state in the database, ensuring it reflects any changes that occurred during the commit.

return to_dict(db_todo):

This line calls the to_dict function (explained in a previous snippet) to convert the db_todo object to a dictionary, and returns this dictionary. This dictionary will be serialized to JSON and sent back to the client as the HTTP response.

DELETE Request – Deleting a TODO

Finally, let’s create an endpoint for deleting a TODO. Add the following code to your `main.py` file:

@app.delete("/todos/{todo_id}", response_model=dict)
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    db_todo = db.query(TodoInDB).filter(TodoInDB.id == todo_id).first()
    if db_todo is None:
        raise HTTPException(status_code=404, detail="Todo not found")
    db.delete(db_todo)
    db.commit()
    return {"message": "Todo deleted successfully"}

This endpoint expects an `id` parameter in the URL and deletes the TODO with the specified `id` if it exists in our list of TODOs.

@app.delete("/todos/{todo_id}", response_model=dict):

This line is a FastAPI decorator that defines a new HTTP DELETE endpoint at the path /todos/{todo_id}. The response_model parameter indicates that the endpoint will return a dictionary. The {todo_id} in the path is a path parameter that will be passed to the delete_todo function.

def delete_todo(todo_id: int, db: Session = Depends(get_db))::

This line declares the delete_todo function which will handle requests to the /todos/{todo_id} endpoint. It accepts two arguments: todo_id of type int, and db of type Session. FastAPI will call the get_db function to obtain a database session and pass it as the db argument, as specified by the Depends keyword.

db_todo = db.query(TodoInDB).filter(TodoInDB.id == todo_id).first():

This line queries the database for a TodoInDB object with the specified todo_id. It uses SQLAlchemy’s query interface to construct and execute the query.

if db_todo is None: and raise HTTPException(status_code=404, detail="Todo not found"):

These lines check if the query returned a TodoInDB object. If not, it raises an HTTPException with a status code of 404 (Not Found) and a detail message of “Todo not found”.

db.delete(db_todo):

This line marks the db_todo object for deletion in the current database session.

db.commit():

This line commits the current database session, which will delete the db_todo object from the database.

return {"message": "Todo deleted successfully"}:

This line returns a dictionary with a message indicating that the todo item was successfully deleted. This dictionary will be serialized to JSON and sent back to the client as the HTTP response.

Running the FastAPI Application

Now that you have created your FastAPI application and defined your models, endpoints, and database configuration, it’s time to run your application and interact with it.

1. Start the FastAPI Application:

Navigate to the directory containing your FastAPI application file (e.g., main.py).
Run the following command to start your FastAPI application:
```
uvicorn main:app --reload --host 0.0.0.0 --port 8000
```
main:app tells Uvicorn to look for an application instance app in a file main.py.
--reload tells Uvicorn to reload the application whenever the code is changed. This is useful during development but should be omitted in a production environment.
--host 0.0.0.0 tells Uvicorn to listen on all public IPs of your server.
--port 8000 specifies the port on which your application will be hosted.

2. Interact with your application:

Below are examples of curl requests for each endpoint in your FastAPI application. Before running these commands, make sure your FastAPI application is running and accessible at http://localhost:8000.

Creating a new Todo item:

curl -X POST "http://localhost:8000/todos" \
-H "accept: application/json" \
-H "Content-Type: application/json" \
-d '{"task":"Buy groceries","completed":false}'

Response:

{
    "id": 1,
    "task": "Buy groceries",
    "completed": false
}

Retrieving all Todo items:

$ curl -X GET "http://localhost:8000/todos" -H "accept: application/json"

Response (assuming the Todo item created in step 1 is the only item):

[
    {
        "id": 1,
        "task": "Buy groceries",
        "completed": false
    }
]

Updating a Todo item (assuming todo_id is 1 and you want to mark it as completed):

$ curl -X PUT "http://localhost:8000/todos/1" \
-H "accept: application/json" \
-H "Content-Type: application/json" \
-d '{"task":"Buy groceries","completed":true}'

Response:

{
    "id": 1,
    "task": "Buy groceries",
    "completed": true
}

Deleting a Todo item (assuming todo_id is 1):

$ curl -X DELETE "http://localhost:8000/todos/1" -H "accept: application/json"

Response:

{
    "message": "Todo deleted successfully"
}

FastAPI also provides an interactive API documentation interface:
- Open your web browser and navigate to http://127.0.0.1:8000/docs.
- Here you can see all the available endpoints of your API, try them out, and see their responses.

3. Stop your application:

When you want to stop your FastAPI application, simply press CTRL + C in the terminal where Uvicorn is running.

4. Check Application Logs:

While your application is running, Uvicorn will display logs in the terminal. These logs provide useful information about the requests being made to your application and any errors that occur.

Generating Documentation

FastAPI provides built-in support for generating documentation using the OpenAPI standard. To access the documentation, run your FastAPI application and navigate to `http://localhost:8000/docs` in your web browser. You will see an interactive documentation page where you can test your endpoints and view detailed information about each endpoint.

Conclusion

Congratulations! You have successfully built a scalable and secure web application with Python and FastAPI. You have learned how to create all endpoints for a TODO application, set up documentation, and create automated tests. FastAPI’s simplicity and performance make it an excellent choice for building robust web applications. Keep exploring FastAPI’s features and continue building amazing applications!

Building Your First API with FastAPI

Introduction

Why FastAPI?

Setting Up Your Environment

Configuring the Database

Defining Database Models

Creating the TODO Application

GET Request – Retrieving All TODOs

POST Request – Creating a New TODO

PUT Request – Updating a TODO

DELETE Request – Deleting a TODO

Running the FastAPI Application

1. Start the FastAPI Application:

2. Interact with your application:

3. Stop your application:

4. Check Application Logs:

Generating Documentation

Conclusion

Related

Leave a Reply Cancel reply