In this tutorial, we are going to create asynchronous API in Python using Flask and Asyncio. After that, we’ll be testing the API as well. But before that let’s understand what asynchronous API is? You might have heard a lot about Asynchronous and Synchronous programming, right?
Asynchronous programming is a method by which we achieve parallel programming. In this, we define multiple threads where they run independently of the main program or function. For example, there’s a code that needs to be executed. So the compiler will start from the top of the file and will go all the way to the bottom of the screen. But, the difference is that while execution it’ll meet some asynchronous functions. These functions will create their threads and will be executed simultaneously with the main function. These asynchronous functions will run independently from the main function.
On the other hand in synchronous programming, the program will start execution from top to bottom. One difference is that it won’t create multiple threads. So everything will be executed line by line unlike asynchronous where two or more functions can be executed at the same time.
Basically, in asynchronous programming, the function can be executed in any order. While in synchronous programming, the functions will be executed in the top to bottom fashion.
How does Asynchronous API work?
Asynchronous API in Python follows the same principle of Asynchronous programming. These APIs don’t return the data immediately. They provide a callback to the client when the results are ready. In the meantime, the client can continue to execute other functions while the API is preparing the results.
These APIs help a lot in applications where elements are independent of each other. For example, you can use these APIs in an e-commerce website homepage. Where you have tiles of different products. Since the products are not dependent on other elements the API can be used to load products and other elements.
There’s one disadvantage as well, these APIs are not suitable where you need an instant response from the API as the latency is the issue here.
Now we have covered a lot about the Asynchronous API. Let’s try to create asynchronous API in Python!
Prerequisites
- Basic knowledge of Python
- Understanding of Asynchronous programming
- Virtualenv installed
Modules that we’ll use
threading
This package is used to introduce multiple threading in the program. When a program is executed, it is viewed as a process and inside a process, there is a thread. When a program is run synchronously, it creates one process under which there is one thread. But when there’s an asynchronous program there is one process but there is more than one thread.
For example, if there’s a program in which there are two async functions. So it’ll create one process and two threads. The threading module helps us to create multiple threads in a python program.
asyncio
This module lets you write concurrent code using async/await syntax. It is used to create async functions in Python. It also comes with libraries to make database connections, add queues, etc. Asyncio also lets you run routines concurrently, control the subprocess, distribute tasks, create and manage event loops.
flask
So all the libraries above helped our code to make our code multithreaded, introduced async/await, etc. We’re creating APIs, right? We need API endpoints also right? This is where the Flask will help us.
Now let’s start with the project setup.
Step -1: Environment setup
Now create a folder `flasksync` and open it. Sounds good right? Flask + Async.
mkdir flasksync
cd flasksyncOnce you’re in the folder let’s initialize out virtual environment:
virtualenv .Once initialized, let’s install all the dependencies:
pip3 install Flask
pip3 install asyncioNow we’re done with the environment setup. Let’s move ahead with the most interesting part i.e. Coding.
Step -2: Creating API Endpoints
Now create a file `api.py` and open it in a code editor.
Let’s start with the import statements. So paste the below code to import all the modules which are required:
import threading
import asyncio
from flask import Flask, jsonifyOnce done let’s try to get the name of the first thread:
print(f"In flask global level: {threading.current_thread().name}")
app = Flask(__name__)The above line of code will try to get the name of the current thread.
After that, we’ll have to create an API endpoint where we can test the Async nature of the API:
@app.route("/test", methods=["GET"])
def index():
print(f"Inside flask function: {threading.current_thread().name}")
asyncio.set_event_loop(asyncio.new_event_loop())
loop = asyncio.get_event_loop()
result = loop.run_until_complete(hello())
return jsonify({"result": result})Code Explanation:
We have defined an API endpoint `/test` where we’ll set an event loop. Once the loop is set we’ll try to get that event by calling the function `get_event_loop()`. The loop will keep on running until a response is returned from the `hello()`. Once the result is ready, we’ll return it in the response.
One point to remember here is that `hello()` is an async function here.
Now we’ll define the async function `hello()`:
async def hello():
await asyncio.sleep(5)
return 1
if __name__ == "__main__":
app.run(host="0.0.0.0", port=4567, debug=False)The function will do nothing just asynchronously wait for 5 seconds and then return 1.
At the end, your code should look something like this:
import threading
import asyncio
from flask import Flask, jsonify
print(f"In flask global level: {threading.current_thread().name}")
app = Flask(__name__)
@app.route("/test", methods=["GET"])
def index():
print(f"Inside flask function: {threading.current_thread().name}")
asyncio.set_event_loop(asyncio.new_event_loop())
loop = asyncio.get_event_loop()
result = loop.run_until_complete(hello())
return jsonify({"result": result})
async def hello():
await asyncio.sleep(5)
return 1
if __name__ == "__main__":
app.run(host="0.0.0.0", port=4567, debug=False)Step -3: Testing the API
Before running the server we have to run one command:
export FLASK_APP=api.pyOnce done, let’s run the server by using command:
flask runNow open your favourite API testing tool and make a GET request to URL `http://127.0.0.1:5000/test`. You’ll have your result ready in 5 seconds:
Final Words
In this article, we covered the functioning of the Asynchronous API and how to create asynchronous API in Python. This was just a gist, you can make different API endpoints and then use them with your application. This will help you improve the user experience a lot as the independent elements now don’t have to wait for the previous API call to fetch the response. You can also check out how we automated our Twitter posts.
