Developing a FastAPI-based API for bidirectional English-Spanish text translation

Published in

Allient

8 min readMar 29, 2023

What is FastAPI?

FastAPI is a modern, high-performance web framework for building web applications in Python 3.6 or higher. It is designed to be easy to use and fast and provides a clear and concise syntax for defining API routes and data models.

Some of the features of this framework include:

High performance: FastAPI uses the ASGI library to provide an asynchronous and scalable architecture that allows it to handle a large number of requests simultaneously with minimal latency.
Runtime typing and type checking: FastAPI uses Python’s type annotation syntax to define API route parameters and automatically validate input data at runtime, helping to reduce errors and provide accurate documentation.
Automatic API documentation: FastAPI uses the OpenAPI standard and JSON Schema to automatically document the API and provide real-time auto-completion suggestions in code editors.
Easy integration: FastAPI is compatible with a wide range of Python tools and libraries, making it easy to integrate with other technologies.

What is Uvicorn?

Uvicorn is a high-speed ASGI (Asynchronous Server Gateway Interface) web server, which can be used to implement Python web applications. Uvicorn is one of the recommended options for running FastAPI applications, although it can also be used with other ASGI-based web applications.

Figure 1. FastAPI — Python (Source: Real Python)

Installing FastAPI and Uvicorn

pip install fastapi uvicorn

Run an API app with Uvicorn

uvicorn main_file:api_instance --reload

Create a project with FastAPI

This post shows how to develop a project with Poetry, Python, and FastAPI, and make use of a Transformers translation model.

Step 1: Starting a new project with Poetry.

poetry new fastapi-sqlmodel-docker

This will create a new project with the name “fastapi-sqlmodel-docker” and you will have a different directory structure within it.

Figure 2. Directory structure of the project

Step 2: Activate virtual environment.

poetry shell

Step 3: Change the python version.

In the pyproject.toml file you need to change the python version to python = “>=3.10,<3.11” so that there is no problem with the tensorflow library.

Step 4: Add required dependencies.

poetry add fastapi uvicorn transformers[sentencepiece] datasets tensorflow python-multipart passlib bcrypt python-dotenv python-jose

This will add the necessary dependencies to develop the related project.

Step 5: Import necessary libraries for the project.

# __init__.py

from fastapi import FastAPI, Depends, HTTPException
from transformers import pipeline
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from UserInDB import UserInDB
from User import User
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Union
from contextlib import asynccontextmanager
import os, dotenv
from jose import JWTError, jwt

Step 6: Load models using the lifespan() function.

# __init__.py

translator_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load the translation models
    translator_models["model1"] = pipeline("translation", model="Helsinki-NLP/opus-mt-en-es")
    translator_models["model2"] = pipeline("translation", model="Helsinki-NLP/opus-mt-es-en")
    yield
    # Clean up the models and release the resources
    translator_models.clear()

An asynchronous context is defined using the @asynccontextmanager decorator. This function will allow the loading of the models when the server starts, this is allowed thanks to “yield” allowing to execute of the models while they are available. The models will be stored in a dictionary named translator_models. At the end of the execution of the server, the memory of the model dictionary is freed.

Step 7: Carry out the authentication process for the use of each API.

First, a local database will be simulated within the code, for this, it is necessary to create the following files:

# User.py

from pydantic import BaseModel
from typing import Union

class User(BaseModel):
    username: str
    full_name: Union[str, None] = None
    email: Union[str, None] = None
    disabled: Union[bool, None] = None

# UserInDB.py

from User import User

class UserInDB(User):
    hashed_password: str

Additionally, a dictionary of users authorized to use the APIS is added.

# __init__.py

# Load environment variables
dotenv.load_dotenv()
SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = "HS256"


authorized_users = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": os.getenv("PASSWORD"),
        "disabled": False,
    }
}

This dictionary can be modified depending on the number of users that you want to have access to each API.

The password of each user and the SECRET_KEY that will allow the creation of a session token will be saved in a .env file for security purposes:

Additionally, it is necessary to create a .gitignore file in which the created .env file will be placed, if the project is going to be uploaded to a web repository like Github.

Then, to perform the authentication process regarding the use of each API it is necessary to create the following functions:

# __init__.py

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def get_user(db, username: str):
    if username in db:
        user_data = db[username]
        return UserInDB(**user_data)
    return []

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def authenticate_user(db, username: str, password: str):
    user = get_user(db, username)
    if not user:
        raise HTTPException(status_code=401, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"})
    if not verify_password(password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"})
    return user

get_user(): takes a database (db) and a username (username) and returns the user data corresponding to the given username in the form of a “UserInDB” object. If the user is not found, it returns an empty list.
verify_password(): Uses the passlib library to verify that the password entered by the user (plain_password) and the hashed password stored in the .env file (hashed_password) match.
authenticate_user(): uses the above two functions to authenticate the user. It takes a database (db), a username (username), and a password (password). It first uses get_user() to get the user data for the given username. If the user is not found, it returns an HTTP error with a 401 status code. If the user is found, it uses verify_password() to verify that the supplied password matches the hashed password stored in the database. If the verification fails, it returns an HTTP error with a 401 status code. If the authentication succeeds, it returns the user data corresponding to the provided username as a UserInDB object.

Step 8: Create a session token and show the current user.

The following functions are defined:

# __init__.py


def create_token(data: dict, time_expire: Union[datetime, None] = None):
    data_copy = data.copy()
    if time_expire is None:
        expires = datetime.utcnow() + timedelta(minutes=15)
    else:
        expires = datetime.utcnow() + time_expire
    data_copy.update({"exp": expires})
    token_jwt = jwt.encode(data_copy, key=SECRET_KEY, algorithm=ALGORITHM)
    return token_jwt

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        token_decode = jwt.decode(token, key=SECRET_KEY, algorithms=[ALGORITHM])
        username = token_decode.get("sub")
        if username == None:
            raise HTTPException(status_code=401, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"})
    except JWTError:
        raise HTTPException(status_code=401, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"})
    
    user = get_user(authorized_users, username)
    if not user:
        raise HTTPException(status_code=401, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"})
    
    return user

def get_user_disable_current(user: User = Depends(get_current_user)):
    if user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return user

create_token(): takes a data dictionary (data) and an expiration time (time_expire), which is an optional datetime object. If the expiration time is not provided, it is set to 15 minutes from the current time. The function then updates the data dictionary with the key “exp”, which contains the token expiration date and time, and uses the jwt library to encode the data into a JWT token using a specified secret key and algorithm. The function returns the JWT token.
get_current_user(): Takes an authentication token (token) as an optional argument that the Depends decorator uses to inject it into the function. The function decodes the token using the same secret key and algorithm as the create_token() function. It then extracts the username from the decoded token and uses the get_user() function to get the user data corresponding to the username. If the user is not found, it returns an HTTP error with a 401 status code. If the authentication is successful, it returns the user data for the given username as a User object.
get_user_disable_current: Use the get_current_user() function to get the user data corresponding to the provided auth token. It then checks if the user has a “disabled” flag set to True. If so, it returns an HTTP error with a 400-status code. If the user is active, it returns the user’s data in the form of a User object.

Step 9: Define functions for translation from English to Spanish and from Spanish to English.

Using the models loaded and stored inside translator_models, we get:

# __init__.py

def model1_en_es(text: str) -> str:
    return translator_models["model1"](text)[0]['translation_text']

def model2_es_en(text: str) -> str:
    return translator_models["model2"](text)[0]['translation_text']

Step 10: Defining routes.

First, you must create an instance of the FastAPI class where you define the lifetime of the application using the lifespan function that was created earlier.

# __init__.py

app = FastAPI(lifespan=lifespan)

The routes that will be defined for the application are:

# __init__.py

oauth2_scheme = OAuth2PasswordBearer("/token")

# Returns a Hello World message.
@app.get("/")
async def root():
    return {"Message": "Hello World"}

# Returns the current logged in user.
@app.get("/users/me")
async def user(user: User = Depends(get_user_disable_current)):
    return {"username:" : user.username, "full_name": user.full_name, "email": user.email, "disabled": user.disabled}

# Given an input text, returns a translation from English to Spanish
@app.get("/translation-en-es/{text}")
async def translation_model1(text: str, token: str = Depends(oauth2_scheme)):
    return {"English to Spanish": model1_en_es(text)}

# Given an input text, returns a translation from Spanish to English
@app.get("/translation-es-en/{text}")
async def translation_model2(text: str, token: str = Depends(oauth2_scheme)):
    return {"Spanish to English": model2_es_en(text)}

# Return authentication token
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(authorized_users, form_data.username, form_data.password)
    access_token_expires = timedelta(minutes=30)
    access_token_jwt = create_token({"sub": user.username}, access_token_expires)

    return {
        "access_token": access_token_jwt,
        "token_type": "bearer",
    }