1. Introduction
In this blog, you will learn how to use async and await to write asynchronous code and improve the performance of your web API. You will also learn how to use async SQL with FastAPI, a modern and fast web framework for building APIs with Python.
Asynchronous programming is a way of writing code that allows multiple tasks to run concurrently without blocking each other. This can improve the responsiveness and scalability of your applications, especially when dealing with network operations or I/O-bound tasks.
Async and await are two keywords in Python that enable you to write asynchronous code in a simple and elegant way. They allow you to define and call functions that can suspend and resume their execution at certain points, without blocking the main thread.
FastAPI is a web framework that supports asynchronous code and provides many features to make your development easier and faster. It is based on the Starlette framework and uses Pydantic for data validation. It also has built-in support for OpenAPI and Swagger UI, which makes it easy to document and test your API.
Async SQL is a way of interacting with relational databases using asynchronous code. It can improve the performance and efficiency of your database operations, as well as reduce the risk of deadlocks and timeouts. You will learn how to use async SQL with FastAPI using the SQL Databases package and the async SQLAlchemy core.
By the end of this blog, you will have a solid understanding of how to use async and await in Python and FastAPI, and how to use async SQL with FastAPI. You will also have a working example of a web API that uses async SQL to perform CRUD operations on a SQLite database.
Are you ready to dive into the world of asynchronous programming with FastAPI? Let’s get started!
2. What is Asynchronous Programming?
Asynchronous programming is a way of writing code that allows multiple tasks to run concurrently without blocking each other. This can improve the responsiveness and scalability of your applications, especially when dealing with network operations or I/O-bound tasks.
But what does it mean to run tasks concurrently and without blocking? And how is it different from the traditional way of writing code, which is called synchronous programming?
To understand the difference, let’s look at an example. Suppose you want to write a web API that fetches data from a remote database and returns it to the client. A typical way of writing this code would be something like this:
@app.get("/data")
def get_data():
# Connect to the database
db = connect_to_database()
# Query the data
data = db.query("SELECT * FROM table")
# Return the data
return data
This code is synchronous, which means that each line of code is executed one after the other, and each line of code waits for the previous one to finish before starting. This also means that if one line of code takes a long time to execute, such as connecting to the database or querying the data, the whole function will be blocked until that line of code is done. This can cause problems if you have many requests coming to your API, as each request will have to wait for the previous one to finish before being processed.
Asynchronous programming solves this problem by allowing you to write code that can suspend and resume its execution at certain points, without blocking the main thread. This way, you can have multiple tasks running at the same time, and switch between them whenever one of them is waiting for something, such as a network response or a disk operation. This can make your code more efficient and responsive, as you can handle more requests in less time.
But how do you write asynchronous code in Python? And how do you use it with FastAPI? That’s what you will learn in the next sections of this blog. Stay tuned!
2.1. The Difference Between Synchronous and Asynchronous Code
In the previous section, you learned what asynchronous programming is and why it is useful for web applications. In this section, you will learn how synchronous and asynchronous code differ from each other, and how you can recognize them in Python.
Synchronous code is the default way of writing code in Python. It means that each line of code is executed one after the other, and each line of code waits for the previous one to finish before starting. This also means that if one line of code takes a long time to execute, such as a network request or a database query, the whole program will be blocked until that line of code is done. This can cause problems if you have multiple tasks that need to run at the same time, or if you want to make your program more responsive and efficient.
Asynchronous code, on the other hand, is a way of writing code that allows multiple tasks to run concurrently without blocking each other. It means that you can define and call functions that can suspend and resume their execution at certain points, without blocking the main thread. This way, you can have multiple tasks running at the same time, and switch between them whenever one of them is waiting for something, such as a network response or a disk operation. This can make your program more efficient and responsive, as you can handle more tasks in less time.
But how can you tell if a code is synchronous or asynchronous in Python? There are two main clues that can help you identify them:
- The async and await keywords. These are two keywords that are used to define and call asynchronous functions in Python. You will learn more about them in the next section, but for now, you can think of them as markers that indicate that a function is asynchronous and that it can suspend and resume its execution. If you see these keywords in a code, it means that the code is asynchronous.
- The asyncio module. This is a module that provides a set of tools and features to write and run asynchronous code in Python. It includes an event loop, which is a mechanism that manages the execution of asynchronous tasks, and a set of coroutines, which are special types of asynchronous functions that can be used with the event loop. You will learn more about the asyncio module in the next section, but for now, you can think of it as a library that enables you to write and run asynchronous code in Python. If you see this module imported in a code, it means that the code is asynchronous.
Here is an example of a synchronous code that fetches data from a remote database and returns it to the client:
@app.get("/data")
def get_data():
# Connect to the database
db = connect_to_database()
# Query the data
data = db.query("SELECT * FROM table")
# Return the data
return data
And here is an example of an asynchronous code that does the same thing, but using the async and await keywords and the asyncio module:
import asyncio
from sql_databases import Database
@app.get("/data")
async def get_data():
# Connect to the database
db = Database("sqlite:///database.db")
await db.connect()
# Query the data
data = await db.fetch_all("SELECT * FROM table")
# Return the data
return data
As you can see, the asynchronous code uses the async and await keywords to define and call the get_data function, and the asyncio module to import the Database class, which is a wrapper for the async SQLAlchemy core. This indicates that the code is asynchronous and that it can run multiple tasks concurrently without blocking.
Now that you know the difference between synchronous and asynchronous code in Python, you might wonder what are the benefits of using asynchronous code over synchronous code. That’s what you will learn in the next section. Keep reading!
2.2. The Benefits of Asynchronous Code
In the previous section, you learned how to distinguish between synchronous and asynchronous code in Python. In this section, you will learn what are the benefits of using asynchronous code over synchronous code, and how it can improve the performance and scalability of your web applications.
One of the main benefits of asynchronous code is that it can handle more tasks in less time, without blocking the main thread. This is because asynchronous code can switch between tasks whenever one of them is waiting for something, such as a network response or a disk operation. This way, you can make use of the idle time of your program, and avoid wasting resources on waiting for something to finish.
Another benefit of asynchronous code is that it can improve the responsiveness of your applications, especially when dealing with network operations or I/O-bound tasks. This is because asynchronous code can handle multiple requests concurrently, without making the client wait for the server to finish one request before processing another. This way, you can provide a better user experience, and avoid timeouts or errors due to slow or unavailable resources.
A third benefit of asynchronous code is that it can simplify the logic and structure of your code, by using the async and await keywords and the asyncio module. This is because asynchronous code can avoid using callbacks, threads, or processes, which can make your code more complex and difficult to maintain. Instead, asynchronous code can use coroutines, which are special types of functions that can suspend and resume their execution, and look like regular synchronous functions. This way, you can write your code in a more readable and elegant way, and avoid common pitfalls of concurrency, such as race conditions or deadlocks.
These are some of the benefits of using asynchronous code over synchronous code in Python. Of course, asynchronous code is not a silver bullet, and it has its own challenges and limitations, such as error handling, debugging, or testing. However, for many use cases, such as web development, asynchronous code can offer significant advantages and make your code more efficient and scalable.
In the next section, you will learn how to use the async and await keywords in Python, which are the core features of asynchronous programming in this language. Stay tuned!
3. How to Use Async and Await in Python
In this section, you will learn how to use the async and await keywords in Python, which are the core features of asynchronous programming in this language. You will also learn how to define and call asynchronous functions, and how to handle errors and exceptions in asynchronous code.
The async and await keywords are two keywords that enable you to write asynchronous code in a simple and elegant way. They allow you to define and call functions that can suspend and resume their execution at certain points, without blocking the main thread. They also make your code look like regular synchronous code, which makes it more readable and maintainable.
The async keyword is used to define an asynchronous function, which is also called a coroutine. A coroutine is a special type of function that can suspend and resume its execution, and return a result or an exception. A coroutine can also await another coroutine, which means that it can suspend its execution until the other coroutine is done, and then resume with the result or the exception of the other coroutine.
The await keyword is used to call an asynchronous function, which is also called awaiting a coroutine. Awaiting a coroutine means that you suspend the execution of the current coroutine until the awaited coroutine is done, and then resume with the result or the exception of the awaited coroutine. Awaiting a coroutine also allows the event loop, which is the mechanism that manages the execution of asynchronous tasks, to run other coroutines while the current one is waiting.
Here is an example of how to use the async and await keywords in Python:
import asyncio
# Define an asynchronous function using the async keyword
async def hello():
# Suspend the execution of the function for 1 second using the asyncio.sleep coroutine
await asyncio.sleep(1)
# Return the result of the function
return "Hello, world!"
# Define another asynchronous function that calls the hello function using the await keyword
async def greet():
# Await the hello function and assign the result to a variable
message = await hello()
# Print the message
print(message)
# Create an event loop
loop = asyncio.get_event_loop()
# Run the greet function using the event loop
loop.run_until_complete(greet())
# Close the event loop
loop.close()
This code defines two asynchronous functions, hello and greet, using the async keyword. The hello function suspends its execution for 1 second using the asyncio.sleep coroutine, and then returns the string “Hello, world!”. The greet function awaits the hello function using the await keyword, and then prints the message. The code also creates an event loop, which is the mechanism that runs the asynchronous tasks, and uses it to run the greet function.
If you run this code, you will see the following output:
Hello, world!
This shows that the code works as expected, and that the greet function waits for the hello function to finish before printing the message.
However, what if something goes wrong in the asynchronous code? How do you handle errors and exceptions in asynchronous code? That’s what you will learn in the next section. Keep reading!
3.1. The async and await Keywords
In this section, you will learn how to use the async and await keywords in Python, which are the core features of asynchronous programming in this language. You will also learn how to define and call asynchronous functions, and how to handle errors and exceptions in asynchronous code.
The async keyword is used to define an asynchronous function, which is also called a coroutine. A coroutine is a special type of function that can suspend and resume its execution, and return a result or an exception. A coroutine can also await another coroutine, which means that it can suspend its execution until the other coroutine is done, and then resume with the result or the exception of the other coroutine.
The await keyword is used to call an asynchronous function, which is also called awaiting a coroutine. Awaiting a coroutine means that you suspend the execution of the current coroutine until the awaited coroutine is done, and then resume with the result or the exception of the awaited coroutine. Awaiting a coroutine also allows the event loop, which is the mechanism that manages the execution of asynchronous tasks, to run other coroutines while the current one is waiting.
Here is an example of how to use the async and await keywords in Python:
import asyncio
# Define an asynchronous function using the async keyword
async def hello():
# Suspend the execution of the function for 1 second using the asyncio.sleep coroutine
await asyncio.sleep(1)
# Return the result of the function
return "Hello, world!"
# Define another asynchronous function that calls the hello function using the await keyword
async def greet():
# Await the hello function and assign the result to a variable
message = await hello()
# Print the message
print(message)
# Create an event loop
loop = asyncio.get_event_loop()
# Run the greet function using the event loop
loop.run_until_complete(greet())
# Close the event loop
loop.close()
This code defines two asynchronous functions, hello and greet, using the async keyword. The hello function suspends its execution for 1 second using the asyncio.sleep coroutine, and then returns the string “Hello, world!”. The greet function awaits the hello function using the await keyword, and then prints the message. The code also creates an event loop, which is the mechanism that runs the asynchronous tasks, and uses it to run the greet function.
If you run this code, you will see the following output:
Hello, world!
This shows that the code works as expected, and that the greet function waits for the hello function to finish before printing the message.
However, what if something goes wrong in the asynchronous code? How do you handle errors and exceptions in asynchronous code? That’s what you will learn in the next section. Keep reading!
3.2. The asyncio Module
In this section, you will learn how to use the asyncio module in Python, which is a module that provides a set of tools and features to write and run asynchronous code in Python. You will also learn how to create and manage an event loop, which is a mechanism that manages the execution of asynchronous tasks, and how to use coroutines, which are special types of asynchronous functions that can be used with the event loop.
The asyncio module is a module that enables you to write and run asynchronous code in Python. It includes an event loop, which is a mechanism that runs the asynchronous tasks, and a set of coroutines, which are special types of asynchronous functions that can be used with the event loop. The asyncio module also provides various utilities and helpers, such as futures, tasks, queues, streams, locks, and timers, that can help you write and manage your asynchronous code.
The event loop is the core feature of the asyncio module. It is a loop that runs in a single thread and executes the asynchronous tasks that are registered with it. The event loop can handle multiple tasks concurrently, by switching between them whenever one of them is waiting for something, such as a network response or a disk operation. The event loop also handles the errors and exceptions that occur in the asynchronous tasks, and provides various callbacks and hooks to monitor and control the execution of the tasks.
To use the event loop, you need to create it, run it, and close it. You can create an event loop using the asyncio.get_event_loop function, which returns the current event loop for the current context. You can run an event loop using the loop.run_until_complete or loop.run_forever methods, which take a coroutine or a future as an argument and run it until it is done or until the loop is stopped. You can close an event loop using the loop.close method, which closes the loop and releases the resources.
Here is an example of how to use the event loop in Python:
import asyncio
# Define an asynchronous function using the async keyword
async def hello():
# Suspend the execution of the function for 1 second using the asyncio.sleep coroutine
await asyncio.sleep(1)
# Return the result of the function
return "Hello, world!"
# Create an event loop
loop = asyncio.get_event_loop()
# Run the hello function using the event loop
result = loop.run_until_complete(hello())
# Print the result
print(result)
# Close the event loop
loop.close()
This code creates an event loop, runs the hello function using the event loop, prints the result, and closes the event loop. If you run this code, you will see the following output:
Hello, world!
This shows that the code works as expected, and that the event loop runs the hello function until it is done.
However, the event loop alone is not enough to write and run asynchronous code in Python. You also need to use coroutines, which are special types of asynchronous functions that can be used with the event loop. That’s what you will learn in the next section. Keep reading!
3.3. The AsyncIO Event Loop
In this section, you will learn how to create and manage an event loop, which is a mechanism that manages the execution of asynchronous tasks in Python. You will also learn how to use the asyncio module, which provides various tools and features to work with the event loop.
The event loop is the core feature of the asyncio module. It is a loop that runs in a single thread and executes the asynchronous tasks that are registered with it. The event loop can handle multiple tasks concurrently, by switching between them whenever one of them is waiting for something, such as a network response or a disk operation. The event loop also handles the errors and exceptions that occur in the asynchronous tasks, and provides various callbacks and hooks to monitor and control the execution of the tasks.
To use the event loop, you need to create it, run it, and close it. You can create an event loop using the asyncio.get_event_loop function, which returns the current event loop for the current context. You can run an event loop using the loop.run_until_complete or loop.run_forever methods, which take a coroutine or a future as an argument and run it until it is done or until the loop is stopped. You can close an event loop using the loop.close method, which closes the loop and releases the resources.
Here is an example of how to create and run an event loop in Python:
import asyncio
# Define an asynchronous function using the async keyword
async def hello():
# Suspend the execution of the function for 1 second using the asyncio.sleep coroutine
await asyncio.sleep(1)
# Return the result of the function
return "Hello, world!"
# Create an event loop
loop = asyncio.get_event_loop()
# Run the hello function using the event loop
result = loop.run_until_complete(hello())
# Print the result
print(result)
# Close the event loop
loop.close()
This code creates an event loop, runs the hello function using the event loop, prints the result, and closes the event loop. If you run this code, you will see the following output:
Hello, world!
This shows that the code works as expected, and that the event loop runs the hello function until it is done.
However, running a single coroutine with the event loop is not very useful. You might want to run multiple coroutines concurrently, and get the results of all of them. You might also want to create and cancel tasks, which are wrappers for coroutines that can be managed by the event loop. You might also want to use various utilities and helpers that the asyncio module provides, such as futures, queues, streams, locks, and timers, that can help you write and manage your asynchronous code.
That’s what you will learn in the next sections of this blog. Stay tuned!
4. How to Use Async and Await in FastAPI
In this section, you will learn how to use the async and await keywords in FastAPI, which is a modern and fast web framework for building APIs with Python. You will also learn how to define and call asynchronous endpoints, and how to use the @app.get decorator and the async def function to create and run your web API.
FastAPI is a web framework that supports asynchronous code and provides many features to make your development easier and faster. It is based on the Starlette framework and uses Pydantic for data validation. It also has built-in support for OpenAPI and Swagger UI, which makes it easy to document and test your API.
FastAPI allows you to write asynchronous code using the async and await keywords, just like you would do in plain Python. You can define and call asynchronous functions, and use the asyncio module and its features to write and manage your asynchronous code. FastAPI also provides a simple and elegant way to create and run your web API using the @app.get decorator and the async def function.
The @app.get decorator is a decorator that registers a function as an endpoint for the GET method of your web API. It takes a path as an argument, which is the URL of the endpoint, and optionally some parameters, such as response_model, status_code, or dependencies, which are used to customize the behavior and the output of the endpoint. The @app.get decorator also supports path and query parameters, which are variables that can be passed in the URL of the endpoint.
The async def function is a function that defines the logic of the endpoint. It takes the same parameters as the @app.get decorator, and returns the response of the endpoint. The async def function can also use the await keyword to call other asynchronous functions, such as database queries or network requests, and suspend and resume its execution without blocking the main thread.
Here is an example of how to use the @app.get decorator and the async def function in FastAPI:
from fastapi import FastAPI
import asyncio
from sql_databases import Database
# Create a FastAPI instance
app = FastAPI()
# Create a database instance
db = Database("sqlite:///database.db")
# Define an endpoint for the GET method using the @app.get decorator
@app.get("/data")
# Define the logic of the endpoint using the async def function
async def get_data():
# Connect to the database using the await keyword
await db.connect()
# Query the data using the await keyword
data = await db.fetch_all("SELECT * FROM table")
# Return the data
return data
This code creates a FastAPI instance, a database instance, and an endpoint for the GET method using the @app.get decorator and the async def function. The endpoint takes the path “/data” as an argument, and returns the data from the database. The async def function uses the await keyword to connect to the database and query the data, and suspends and resumes its execution without blocking the main thread.
If you run this code, you will have a web API that can handle GET requests to the “/data” endpoint, and return the data from the database. You can also use the OpenAPI and Swagger UI features of FastAPI to document and test your API.
Now that you know how to use async and await in FastAPI, you might wonder how to use async SQL with FastAPI, which is a way of interacting with relational databases using asynchronous code. That’s what you will learn in the next section. Keep reading!
4.1. The FastAPI Framework
FastAPI is a web framework that supports asynchronous code and provides many features to make your development easier and faster. It is based on the Starlette framework and uses Pydantic for data validation. It also has built-in support for OpenAPI and Swagger UI, which makes it easy to document and test your API.
To use FastAPI, you need to install it using pip:
pip install fastapi
You also need to install an ASGI server, such as uvicorn, to run your app:
pip install uvicorn
Then, you can create a file called main.py and write your first FastAPI app:
# Import FastAPI
from fastapi import FastAPI
# Create an app instance
app = FastAPI()
# Define a route
@app.get("/")
def hello():
# Return a JSON response
return {"message": "Hello, world!"}
To run your app, use the following command:
uvicorn main:app --reload
This will start the server on http://127.0.0.1:8000. You can visit this URL in your browser and see the JSON response. You can also visit http://127.0.0.1:8000/docs to see the interactive documentation generated by FastAPI, where you can test your API endpoints.
As you can see, FastAPI is very simple and intuitive to use. You just need to create an app instance, define routes using decorators, and return JSON responses. FastAPI takes care of the rest, such as validating the data, generating the documentation, and handling the requests and responses.
But how do you use async and await with FastAPI? And how do you use async SQL with FastAPI? That’s what you will learn in the next sections of this blog. Keep reading!
4.2. The @app.get Decorator
The @app.get decorator is one of the ways to define routes in FastAPI. It tells FastAPI that the function below it should handle GET requests to the specified path. For example, the following code defines a route for the path “/data” that returns some data:
@app.get("/data")
def get_data():
# Return some data
return {"data": [1, 2, 3]}
The @app.get decorator also supports parameters, such as path parameters, query parameters, and request body. You can use them to pass data from the client to the server. For example, the following code defines a route for the path “/data/{id}” that takes an integer id as a path parameter and returns the corresponding data:
@app.get("/data/{id}")
def get_data(id: int):
# Return the data with the given id
return {"data": id}
You can also use the @app.get decorator to define the response model, the status code, the tags, and the summary of your route. These are optional arguments that help FastAPI to generate the documentation and validate the data. For example, the following code defines a route for the path “/user” that takes a username as a query parameter, returns a User object as the response model, sets the status code to 200, adds the tag “user”, and provides a summary of the route:
from pydantic import BaseModel
# Define the User model
class User(BaseModel):
name: str
age: int
email: str
@app.get("/user", response_model=User, status_code=200, tags=["user"], summary="Get a user by username")
def get_user(username: str):
# Return a user object with the given username
return User(name=username, age=25, email=f"{username}@example.com")
As you can see, the @app.get decorator is very powerful and flexible. It allows you to define routes in a simple and declarative way, and FastAPI takes care of the rest. But how do you use async and await with the @app.get decorator? And how do you use async SQL with the @app.get decorator? That’s what you will learn in the next sections of this blog. Don’t miss them!
4.3. The async def Function
The async def function is another way to use async and await with FastAPI. It allows you to define a function that can run asynchronously and use the await keyword to call other asynchronous functions. For example, the following code defines an async def function that fetches data from a remote API and returns it:
import httpx
# Define an async def function
@app.get("/fetch")
async def fetch_data():
# Use the await keyword to call an asynchronous function
response = await httpx.get("https://example.com/api/data")
# Return the data from the response
return response.json()
The async def function is useful when you need to perform some network operations or I/O-bound tasks that can take a long time to complete. By using the await keyword, you can suspend the execution of the function until the result is ready, without blocking the main thread. This way, you can handle more requests in less time and improve the performance of your web API.
However, the async def function is not suitable for CPU-bound tasks that require a lot of computation. In that case, you should use a normal def function and run it in a separate thread or process using the asyncio module. You will learn more about the asyncio module in the next section of this blog.
As you can see, the async def function is very easy and convenient to use. You just need to add the async keyword before the def keyword, and use the await keyword to call other asynchronous functions. FastAPI will automatically detect the async def function and run it in an asynchronous way. But how do you use async SQL with the async def function? That’s what you will learn in the next sections of this blog. Stay tuned!
5. How to Use Async SQL with FastAPI
Async SQL is a way of interacting with relational databases using asynchronous code. It can improve the performance and efficiency of your database operations, as well as reduce the risk of deadlocks and timeouts. To use async SQL with FastAPI, you need to install the SQL Databases package, which provides a simple and elegant way to use async SQLAlchemy core with FastAPI.
To install the SQL Databases package, use pip:
pip install databases
Then, you need to create a database connection object, using the URL of your database. For example, the following code creates a connection object for a SQLite database:
from databases import Database
# Create a database connection object
database = Database("sqlite:///example.db")
You also need to create a metadata object, which is a collection of Table objects that describe your database schema. For example, the following code creates a metadata object with one table called “users”:
from sqlalchemy import MetaData, Table, Column, Integer, String
# Create a metadata object
metadata = MetaData()
# Create a table object
users = Table(
"users",
metadata,
Column("id", Integer, primary_key=True),
Column("name", String),
Column("age", Integer),
Column("email", String),
)
Next, you need to connect and disconnect the database in your app startup and shutdown events. For example, the following code adds two event handlers to your app instance:
@app.on_event("startup")
async def startup():
# Connect to the database
await database.connect()
@app.on_event("shutdown")
async def shutdown():
# Disconnect from the database
await database.disconnect()
Now, you are ready to use async SQL with FastAPI. You can use the database object to execute queries and fetch results using the await keyword. For example, the following code defines a route for the path “/users” that returns all the users from the database:
@app.get("/users", response_model=List[User])
async def get_users():
# Execute a query
query = users.select()
# Fetch the results
results = await database.fetch_all(query)
# Return the results
return results
As you can see, using async SQL with FastAPI is very simple and intuitive. You just need to install the SQL Databases package, create a database connection object and a metadata object, connect and disconnect the database in your app events, and use the await keyword to execute queries and fetch results. FastAPI will automatically handle the requests and responses, and validate the data using the response model.
But how do you perform CRUD operations with async SQL and FastAPI? And how do you use async SQL with the async def function? That’s what you will learn in the next sections of this blog. Don’t miss them!
5.1. The SQL Databases Package
The SQL Databases package is a Python library that provides a simple and elegant way to use async SQLAlchemy core with FastAPI. SQLAlchemy core is a low-level SQL toolkit that allows you to write SQL queries using Python objects. It is different from SQLAlchemy ORM, which is a high-level abstraction that maps Python classes to database tables. The SQL Databases package supports both SQLite and PostgreSQL databases, and allows you to execute queries and fetch results using the await keyword.
To use the SQL Databases package, you need to install it using pip:
pip install databases
Then, you need to import the Database class from the databases module, and create a database connection object using the URL of your database. For example, the following code creates a connection object for a SQLite database:
from databases import Database
# Create a database connection object
database = Database("sqlite:///example.db")
You also need to import the MetaData class from the sqlalchemy module, and create a metadata object that describes your database schema. A metadata object is a collection of Table objects that define the columns, types, constraints, and indexes of your tables. For example, the following code creates a metadata object with one table called “users”:
from sqlalchemy import MetaData, Table, Column, Integer, String
# Create a metadata object
metadata = MetaData()
# Create a table object
users = Table(
"users",
metadata,
Column("id", Integer, primary_key=True),
Column("name", String),
Column("age", Integer),
Column("email", String),
)
Next, you need to connect and disconnect the database in your app startup and shutdown events. This ensures that the database connection is established and closed properly when your app runs. You can use the on_event decorator to add event handlers to your app instance. For example, the following code adds two event handlers that connect and disconnect the database:
@app.on_event("startup")
async def startup():
# Connect to the database
await database.connect()
@app.on_event("shutdown")
async def shutdown():
# Disconnect from the database
await database.disconnect()
Now, you are ready to use the SQL Databases package with FastAPI. You can use the database object to execute queries and fetch results using the await keyword. The database object supports various methods, such as execute, fetch_one, fetch_all, and iterate. For example, the following code defines a route for the path “/users” that returns all the users from the database:
@app.get("/users", response_model=List[User])
async def get_users():
# Execute a query
query = users.select()
# Fetch the results
results = await database.fetch_all(query)
# Return the results
return results
As you can see, using the SQL Databases package with FastAPI is very simple and intuitive. You just need to install the package, create a database connection object and a metadata object, connect and disconnect the database in your app events, and use the await keyword to execute queries and fetch results. FastAPI will automatically handle the requests and responses, and validate the data using the response model.
But how do you perform CRUD operations with the SQL Databases package and FastAPI? And how do you use the SQL Databases package with the async def function? That’s what you will learn in the next sections of this blog. Stay tuned!
5.2. The Async SQLAlchemy Core
The Async SQLAlchemy Core is a part of the SQLAlchemy library that allows you to write SQL queries using Python objects and execute them asynchronously. SQLAlchemy Core is a low-level SQL toolkit that gives you more control and flexibility over your database operations. It is different from SQLAlchemy ORM, which is a high-level abstraction that maps Python classes to database tables. The Async SQLAlchemy Core supports both SQLite and PostgreSQL databases, and works well with the SQL Databases package and FastAPI.
To use the Async SQLAlchemy Core, you need to install it using pip:
pip install sqlalchemy
Then, you need to import the MetaData class from the sqlalchemy module, and create a metadata object that describes your database schema. A metadata object is a collection of Table objects that define the columns, types, constraints, and indexes of your tables. For example, the following code creates a metadata object with one table called “users”:
from sqlalchemy import MetaData, Table, Column, Integer, String
# Create a metadata object
metadata = MetaData()
# Create a table object
users = Table(
"users",
metadata,
Column("id", Integer, primary_key=True),
Column("name", String),
Column("age", Integer),
Column("email", String),
)
Next, you need to import the create_engine function from the sqlalchemy module, and create an engine object that connects to your database. An engine object is responsible for managing the connection pool and executing the queries. You need to pass the URL of your database and the echo parameter, which controls the logging level. For example, the following code creates an engine object for a SQLite database:
from sqlalchemy import create_engine
# Create an engine object
engine = create_engine("sqlite+aiosqlite:///example.db", echo=True)
Now, you are ready to use the Async SQLAlchemy Core with FastAPI. You can use the engine object to execute queries and fetch results using the await keyword. The engine object supports various methods, such as execute, scalar, fetchone, fetchall, and fetchmany. For example, the following code defines a route for the path “/users” that returns all the users from the database:
@app.get("/users", response_model=List[User])
async def get_users():
# Execute a query
query = users.select()
# Fetch the results
results = await engine.fetch_all(query)
# Return the results
return results
As you can see, using the Async SQLAlchemy Core with FastAPI is very simple and intuitive. You just need to install the library, create a metadata object and an engine object, and use the await keyword to execute queries and fetch results. FastAPI will automatically handle the requests and responses, and validate the data using the response model.
But how do you perform CRUD operations with the Async SQLAlchemy Core and FastAPI? And how do you use the Async SQLAlchemy Core with the async def function? That’s what you will learn in the next sections of this blog. Stay tuned!
5.3. The Async Database Connection
The Async Database Connection is a feature of the SQL Databases package that allows you to use the async SQLAlchemy core with FastAPI. It is a wrapper around the engine object that provides a simple and consistent interface for executing queries and fetching results using the await keyword. It also handles the connection pool and the transaction management for you.
To use the Async Database Connection, you need to import the Database class from the databases module, and create a database connection object using the URL of your database. For example, the following code creates a connection object for a SQLite database:
from databases import Database
# Create a database connection object
database = Database("sqlite+aiosqlite:///example.db")
You also need to connect and disconnect the database in your app startup and shutdown events. This ensures that the database connection is established and closed properly when your app runs. You can use the on_event decorator to add event handlers to your app instance. For example, the following code adds two event handlers that connect and disconnect the database:
@app.on_event("startup")
async def startup():
# Connect to the database
await database.connect()
@app.on_event("shutdown")
async def shutdown():
# Disconnect from the database
await database.disconnect()
Now, you are ready to use the Async Database Connection with FastAPI. You can use the database object to execute queries and fetch results using the await keyword. The database object supports various methods, such as execute, fetch_one, fetch_all, and iterate. For example, the following code defines a route for the path “/users” that returns all the users from the database:
@app.get("/users", response_model=List[User])
async def get_users():
# Execute a query
query = users.select()
# Fetch the results
results = await database.fetch_all(query)
# Return the results
return results
As you can see, using the Async Database Connection with FastAPI is very simple and intuitive. You just need to import the Database class, create a database connection object, connect and disconnect the database in your app events, and use the await keyword to execute queries and fetch results. FastAPI will automatically handle the requests and responses, and validate the data using the response model.
But how do you perform CRUD operations with the Async Database Connection and FastAPI? And how do you use the Async Database Connection with the async def function? That’s what you will learn in the next sections of this blog. Stay tuned!
6. Conclusion
In this blog, you have learned how to use async and await to write asynchronous code and improve the performance of your web API. You have also learned how to use async SQL with FastAPI, using the SQL Databases package and the async SQLAlchemy core. You have seen how to create a database connection object and a metadata object, how to connect and disconnect the database in your app events, how to execute queries and fetch results using the await keyword, and how to perform CRUD operations with the async def function. You have also seen how FastAPI handles the requests and responses, and validates the data using the response model.
Asynchronous programming is a powerful and elegant way of writing code that can handle multiple tasks concurrently without blocking each other. It can make your code more efficient and responsive, especially when dealing with network operations or I/O-bound tasks. Async and await are two keywords in Python that enable you to write asynchronous code in a simple and intuitive way. They allow you to define and call functions that can suspend and resume their execution at certain points, without blocking the main thread.
FastAPI is a web framework that supports asynchronous code and provides many features to make your development easier and faster. It is based on the Starlette framework and uses Pydantic for data validation. It also has built-in support for OpenAPI and Swagger UI, which makes it easy to document and test your API. FastAPI allows you to use async and await in your routes and functions, and handles the requests and responses automatically.
Async SQL is a way of interacting with relational databases using asynchronous code. It can improve the performance and efficiency of your database operations, as well as reduce the risk of deadlocks and timeouts. The SQL Databases package and the async SQLAlchemy core are two libraries that allow you to use async SQL with FastAPI. They provide a simple and consistent interface for executing queries and fetching results using the await keyword. They also handle the connection pool and the transaction management for you.
By the end of this blog, you have gained a solid understanding of how to use async and await in Python and FastAPI, and how to use async SQL with FastAPI. You have also created a working example of a web API that uses async SQL to perform CRUD operations on a SQLite database. You can find the complete code for this blog on GitHub.
We hope you have enjoyed this blog and learned something new and useful. If you have any questions or feedback, please feel free to leave a comment below. Thank you for reading!
