In this post, we will see how to call an external API from our API.
We will use FastAPI for our API instead, for the external API, we will use Json Server.
We have already seen FastAPI and Json Server in these two post:
Web API – Json Server
Python – Fast API
In details, I want to create a FastAPI where in input (using the POST) I take an User object and then I will pass this object in the external API to add this new object to the list of User already existing in the API.
We start to define the data.json for the external API:
[DATA.JSON]
{
"users": [
{
"id": "1",
"username": "username1",
"password": "password1"
},
{
"id": "2",
"username": "username2",
"password": "password2"
}
]
}
Then, we start the API using the command:
json-server --watch data.json
Finally, with a client API we can verify that it works:
Now, we will create the FastAPI and we start to define the User entity:
[USER.PY]
from pydantic import BaseModel
class UserBase(BaseModel):
id: int
username: str
password: str
Then, we define our API and the endpoint used to take an User object:
[MAIN.PY]
# Import FastAPI to create a web application,
# HTTPException to handle HTTP errors, and the
# status module for status codes.
from fastapi import FastAPI, HTTPException, status
# Import UserBase class from the User module.
from User import UserBase
app = FastAPI()
# Define a route for a POST request on the "/user" endpoint.
# The function is expected to return a 200 OK status upon successful execution.
@app.post("/user", status_code=status.HTTP_200_OK)
# Define an asynchronous function to insert a user.
async def insert_user(user: UserBase):
try:
# Attempt to serialize the user object.
# This method 'model_dump' is assumed to return a dictionary or
# JSON-like structure representing the user.
data = user.model_dump()
print(data)
except Exception as e:
# If any exception occurs during the process,
# raise an HTTPException with a 400 Bad Request status.
raise HTTPException(status_code=400, detail=str(e))
Finally, we run the API and then, with a API client we try to call the endpoint; the result, will be the object passed in input:
py.exe -m uvicorn main:app --reload
Now, we will add a method called ‘send_user_to_add’ that we will use to call the external API to add the new user but, before, we have to install the library httpx using the command:
pip install httpx
I just want to say that httpx is a fully featured HTTP client for Python, similar to the well-known request library but with additional capabilities.
# Define an asynchronous function to send user data to an external service
async def send_user_to_add(user):
# The URL of the external service endpoint
url = "http://127.0.0.1:3000/users"
# Create an asynchronous HTTP client context.
# This context manager ensures that the client is properly closed after the request.
async with httpx.AsyncClient() as client:
try:
# Attempt to post the user data to the specified URL
response = await client.post(url, json=user)
except httpx.RequestError as exc:
# This block catches network-related errors (e.g., DNS failures, refused connections).
# It raises an HTTPException, which is a way to signal an HTTP error from a FastAPI application.
raise HTTPException(
status_code=400, detail=f"Request to external API failed: {str(exc)}"
)
except httpx.HTTPStatusError as exc:
# This block catches cases where the external API responds with a 4xx or 5xx error status code.
# It raises an HTTPException with the status code and detail from the external API response.
raise HTTPException(
status_code=exc.response.status_code,
detail=f"Error response from external API"
)
Finally, we modify the ‘insert_user’ method in order to use the method ‘send_user_to_add’:
@app.post("/user", status_code=status.HTTP_200_OK)
# Define an asynchronous function to insert a user.
async def insert_user(user: UserBase):
try:
# Attempt to serialize the user object.
# This method 'model_dump' is assumed to return a dictionary or
# JSON-like structure representing the user.
data = user.model_dump()
await send_user_to_add(data)
return {"message": "Request received successfully"}
except Exception as e:
# If any exception occurs during the process,
# raise an HTTPException with a 400 Bad Request status.
raise HTTPException(status_code=400, detail=str(e))
Now, if we run the API, the following will be the result:
The last thing that I want to show, is the BackgroundTasks.
In a nutshell, it is a way to run non-critical operations in the background in a request-response life cycle. It’s a part of the Starlette web framework and, by extension, is well-supported and frequently utilized in the FastAPI Python web application framework.
In this case, I want to use the BackgroundTasks to call the method ‘insert_user’ and the client will receive an HTTP 200 response without having to wait for the completion of it.
This is the complete file:
[MAIN.PY]
from fastapi import FastAPI, HTTPException, BackgroundTasks, status
# Import UserBase class from the User module.
from User import UserBase
import asyncio
import httpx
app = FastAPI()
# Define a route for a POST request on the "/user" endpoint.
# The function is expected to return a 200 OK status upon successful execution.
@app.post("/user", status_code=status.HTTP_200_OK)
# Define an asynchronous function to insert a user.
async def insert_user(background_tasks: BackgroundTasks, user: UserBase):
try:
# Attempt to serialize the user object.
# This method 'model_dump' is assumed to return a dictionary or
# JSON-like structure representing the user.
data = user.model_dump()
# This adds a new background task for execution.
# The `send_user_to_add` function is queued up to run with `data` as its argument.
# The FastAPI's BackgroundTasks manager will handle the operation of the function asynchronously.
# This approach allows the server to respond to the client request immediately, without waiting for this task to complete
background_tasks.add_task(send_user_to_add, data)
return {"message": "Request received successfully"}
except Exception as e:
# If any exception occurs during the process,
# raise an HTTPException with a 400 Bad Request status.
raise HTTPException(status_code=400, detail=str(e))
# Define an asynchronous function to send user data to an external service
async def send_user_to_add(user):
# The URL of the external service endpoint
url = "http://127.0.0.1:3000/users"
# Create an asynchronous HTTP client context.
# This context manager ensures that the client is properly closed after the request.
async with httpx.AsyncClient() as client:
try:
# Wait for 3 seconds before sending the request
await asyncio.sleep(3)
# Attempt to post the user data to the specified URL.
response = await client.post(url, json=user)
except httpx.RequestError as exc:
# This block catches network-related errors (e.g., DNS failures, refused connections).
# It raises an HTTPException, which is a way to signal an HTTP error from a FastAPI application.
raise HTTPException(
status_code=400, detail=f"Request to external API failed: {str(exc)}"
)
except httpx.HTTPStatusError as exc:
# This block catches cases where the external API responds with a 4xx or 5xx error status code.
# It raises an HTTPException with the status code and detail from the external API response.
raise HTTPException(
status_code=exc.response.status_code,
detail=f"Error response from external API"
)
If we run the API, the following will be the result:
