FastAPI with ***** ? Softwares and Servers use with FastAPI

FastAPI with ***** ?

softwares that are usally use with FastAPI

FastAPI is a versatile framework that can be integrated with various software and tools to enhance its functionality. Here are some popular ones:

Databases

• SQLAlchemy: An ORM (Object-Relational Mapper) for working with SQL databases.
• Tortoise-ORM: Another ORM that supports asynchronous operations.
• MongoDB: A NoSQL database that can be used with FastAPI through libraries like Motor.

Authentication

• OAuth2: FastAPI has built-in support for OAuth2, which is commonly used for authentication.
• JWT (JSON Web Tokens): Libraries like PyJWT can be used for token-based authentication.

Frontend Frameworks

• React: A popular JavaScript library for building user interfaces.
• Vue.js: Another JavaScript framework for building UIs.
• Angular: A TypeScript-based framework for building web applications.

Deployment

• Docker: Containerization platform to package and deploy FastAPI applications.
• Kubernetes: Orchestration tool for managing containerized applications.
• Heroku: A cloud platform that supports FastAPI deployment.

Testing

• Pytest: A testing framework for writing and running tests.
• HTTPX: An HTTP client for making requests during testing.

Documentation

• Swagger UI: Automatically generated interactive API documentation.
• ReDoc: Another tool for generating API documentation.

Other Tools

• Celery: A task queue for handling background tasks.
• Redis: An in-memory data structure store, often used as a cache or message broker.

 

Servers that are using with FastAPI

Common Server Environments for FastAPI

1. Uvicorn: An ASGI server that is commonly used to run FastAPI applications. It's lightweight and efficient.
2. Daphne: Another ASGI server that can be used with FastAPI, often chosen for its support of WebSocket protocols.
3. Gunicorn: A WSGI server that can be used with FastAPI applications by running them through Uvicorn workers.
4. Nginx: Often used as a reverse proxy server to handle incoming HTTP requests and forward them to the FastAPI application running on an ASGI server like Uvicorn.
5. Docker: As mentioned earlier, Docker can be used to containerize FastAPI applications for easier deployment and scaling on various cloud platforms.

Deployment Platforms

• AWS (Amazon Web Services): FastAPI applications can be deployed using services like EC2 (Elastic Compute Cloud) or ECS (Elastic Container Service) with Docker.
• Google Cloud Platform: You can use Google App Engine, Google Kubernetes Engine, or Compute Engine to deploy FastAPI applications.
• Microsoft Azure: Azure App Service, Azure Kubernetes Service, and Azure Virtual Machines are common choices for deploying FastAPI applications.
• Heroku: A user-friendly platform for deploying applications, including FastAPI, with minimal configuration.

 

 

To adjust docker and Uvicorn to code

app/main.py

from fastapi import FastAPI
from app.routes import predict, train, update, hyperparameter_tuning
from app.utils.error_handler import custom_error_handler
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# Include the routers from different endpoints
app.include_router(predict.router)
app.include_router(train.router)
app.include_router(update.router)
app.include_router(hyperparameter_tuning.router)

# Add error handler
app.add_exception_handler(RequestValidationError, custom_error_handler)
app.add_exception_handler(Exception, custom_error_handler)

# The following lines are added to run the app with Uvicorn
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)  

 

 

Location of the Dockerfile

 

 

Add Authentication

At first I tried the put OAuth2 function with 2FA(two step authentication). The container has not been changed.

app/main.py

from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from datetime import datetime, timedelta
import random
import string

app = FastAPI()

# Secret key to encode the JWT
SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Password hashing context
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# In-memory store for 2FA codes
two_factor_auth_codes = {}

# Fake user database
fake_users_db = {
    "user@example.com": {
        "username": "user@example.com",
        "hashed_password": pwd_context.hash("password"),
    }
}

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

class User(BaseModel):
    username: str

class UserInDB(User):
    hashed_password: str

class TwoFactorData(BaseModel):
    username: str
    code: str

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

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

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def generate_2fa_code():
    return ''.join(random.choices(string.digits, k=6))

def send_2fa_code(username: str, code: str):
    # In a real application, send the code via SMS, email, or an authenticator app
    print(f"2FA code for {username}: {code}")

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=400,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    code = generate_2fa_code()
    two_factor_auth_codes[form_data.username] = code
    send_2fa_code(form_data.username, code)
    return {"access_token": "2fa_required", "token_type": "bearer"}

@app.post("/2fa", response_model=Token)
async def verify_2fa_code(data: TwoFactorData):
    if data.username in two_factor_auth_codes and two_factor_auth_codes[data.username] == data.code:
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"sub": data.username}, expires_delta=access_token_expires
        )
        del two_factor_auth_codes[data.username]  # Remove used code
        return {"access_token": access_token, "token_type": "bearer"}
    else:
        raise HTTPException(
            status_code=400,
            detail="Invalid 2FA code",
            headers={"WWW-Authenticate": "Bearer"},
        )

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

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

# Include the routers from different endpoints
from app.routes import predict, train, update, hyperparameter_tuning
app.include_router(predict.router)
app.include_router(train.router)
app.include_router(update.router)
app.include_router(hyperparameter_tuning.router)

# Add error handler
from app.utils.error_handler import custom_error_handler
from fastapi.exceptions import RequestValidationError
app.add_exception_handler(RequestValidationError, custom_error_handler)
app.add_exception_handler(Exception, custom_error_handler)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)  # This line is added

 

 

app/utils/error_handler.py

from fastapi.responses import JSONResponse
from fastapi import Request

async def custom_error_handler(request: Request, exc):
    return JSONResponse(
        status_code=400,
        content={"message": f"Oops! Something went wrong: {str(exc)}"}
    )

 

 

app/utils/transformers_utils.py

from transformers import ElectraTokenizer, ElectraForSequenceClassification, Trainer, TrainingArguments
import torch
from datasets import load_metric, Dataset
import optuna

class TransformersModel:
    def __init__(self, model_name: str):
        self.tokenizer = ElectraTokenizer.from_pretrained(model_name)
        self.model = ElectraForSequenceClassification.from_pretrained(model_name)
        self.metric = load_metric("accuracy")

    def predict(self, text: str):
        inputs = self.tokenizer(text, return_tensors='pt')
        outputs = self.model(**inputs)
        logits = outputs.logits
        probabilities = torch.nn.functional.softmax(logits, dim=-1)
        predicted_class = torch.argmax(probabilities, dim=1).item()
        return {"class": predicted_class, "probabilities": probabilities.tolist()}

    def train(self, train_dataset: Dataset, eval_dataset: Dataset, output_dir: str, training_args: dict):
        args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=training_args["num_train_epochs"],
            per_device_train_batch_size=training_args["per_device_train_batch_size"],
            per_device_eval_batch_size=training_args["per_device_eval_batch_size"],
            evaluation_strategy="steps",
            save_steps=10_000,
            eval_steps=500,
            logging_dir=f"{output_dir}/logs",
            learning_rate=training_args["learning_rate"],
            weight_decay=training_args["weight_decay"],
        )
        trainer = Trainer(
            model=self.model,
            args=args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            tokenizer=self.tokenizer,
            compute_metrics=self.compute_metrics,
        )
        trainer.train()
        trainer.save_model(output_dir)
        return {"status": "Training completed"}

    def update(self, new_train_texts: list, new_train_labels: list, output_dir: str):
        new_train_dataset = Dataset.from_dict({"text": new_train_texts, "label": new_train_labels})
        return self.train(new_train_dataset, new_train_dataset, output_dir, self.default_training_args())
    
    def hyperparameter_tuning(self, train_dataset: Dataset, eval_dataset: Dataset, output_dir: str):
        def objective(trial):
            training_args = {
                "num_train_epochs": trial.suggest_int("num_train_epochs", 1, 5),
                "per_device_train_batch_size": trial.suggest_categorical("per_device_train_batch_size", [8, 16, 32]),
                "per_device_eval_batch_size": trial.suggest_categorical("per_device_eval_batch_size", [8, 16, 32]),
                "learning_rate": trial.suggest_loguniform("learning_rate", 1e-5, 1

 

 

How to measure the app?

When I use the computer or app. If It is slow I don’t use it. So I know how much important the app speed is.

Let’s check it with Copilot!

The speed of your application might be impacted by the added complexity and features, but there are several strategies you can use to maintain or even improve performance:

Optimization Tips

1. Asynchronous Operations: Ensure that you're making use of FastAPI's asynchronous capabilities (e.g., async/await) to handle I/O-bound tasks like database queries and external API calls efficiently.
2. Efficient Data Processing: Optimize your data processing pipelines to reduce bottlenecks. Use efficient data structures and algorithms, and consider batch processing if applicable.
3. Caching: Implement caching for frequently accessed data or computationally expensive operations. Tools like Redis can be used for caching.
4. Database Optimization: Ensure that your database queries are optimized. Use indexing, pagination, and avoid unnecessary queries.
5. Load Balancing: Distribute the load across multiple instances of your application using load balancers to improve scalability and performance.
6. Profiling and Monitoring: Use profiling tools to identify performance bottlenecks and optimize them. Tools like Prometheus and Grafana can help with monitoring performance metrics.
7. Hardware Resources: Ensure that the server or infrastructure you're using has sufficient hardware resources (CPU, memory, etc.) to handle the increased load.

Assessing Performance

To measure the performance impact of your changes, you can use benchmarking and monitoring tools:

1. Benchmarking Tools: Use tools like ab (Apache Bench), wrk, or locust to simulate load on your application and measure response times and throughput.
2. Profiling Tools: Use profiling tools like cProfile, py-spy, or perf to identify performance bottlenecks in your code.
3. Monitoring Tools: Use monitoring tools like Prometheus, Grafana, or New Relic to track performance metrics in real time.

Thank you for reading! xoxo

 

내가 노션에 정리한 내용을 블로그로 다시 옮겨왔다.

https://snow-light-ac7.notion.site/FastAPI-1989a232cc3580fb9c25ee5693abf945#19d9a232cc3580d0a54cf27ff2ba7615

FastAPI | Notion