ANI

Application for testing to press Fastapi – Kdnugget

nimda August 15, 2025

0 5 6 minutes read

Application for testing to press Fastapi – Kdnugget

Photo by the writer

Obvious Introduction

Pressure checks is important to understand how your app behaves under heavy loads. For Machine Learning-Powed-powered apis, it is very important because the model tendency can p. By implementing a large number of users, we can see operating bottles, determine the power of our system, and ensure reliability.

In this lesson, we will use it:

Fastapi: Modern, fast (high-level) web site for building APIs with Python.
Vicorn: ASGI server to use our Fastapi app.
Locusts: Open source test tool. It describes user conduct with the Python Code, and jump your system for employees at the same time.
Skikit-Read: With our model of the examples of model.

Obvious 1. Projects and dependency

Set the project structure and enter the required dependence.

Cause requirements.txt The file and add the following Python packages:

fastapi==0.115.12
locust==2.37.10
numpy==2.3.0
pandas==2.3.0
pydantic==2.11.5
scikit-learn==1.7.0
uvicorn==0.34.3
orjson==3.10.18

Open your terminal, create visual nature, and use it.

python -m venv venv
venvScriptsactivate

Enter all Python packages using requirements.txt file.

pip install -r requirements.txt

Obvious 2. Creating a Fastapi Request

In this stage, we will create a training file for the return model, Pydantic models, and Fastapi application.

This ml_model.py Treats a machine learning model. Using a singleton pattern to ensure only one example of the loaded model. The model is a rarressor of random forest trained for California housing dataset. If the previously trained model (model.Pkl and scarder.pkl) is not available, train and save new.

app/ml_model.py:

import os
import threading

import joblib
import numpy as np
from sklearn.datasets import fetch_california_housing
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

class MLModel:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if not hasattr(self, "initialized"):
            self.model = None
            self.scaler = None
            self.model_path = "model.pkl"
            self.scaler_path = "scaler.pkl"
            self.feature_names = None
            self.initialized = True
            self.load_or_create_model()

    def load_or_create_model(self):
        """Load existing model or create a new one using California housing dataset"""
        if os.path.exists(self.model_path) and os.path.exists(self.scaler_path):
            self.model = joblib.load(self.model_path)
            self.scaler = joblib.load(self.scaler_path)
            housing = fetch_california_housing()
            self.feature_names = housing.feature_names
            print("Model loaded successfully")
        else:
            print("Creating new model...")
            housing = fetch_california_housing()
            X, y = housing.data, housing.target
            self.feature_names = housing.feature_names

            X_train, X_test, y_train, y_test = train_test_split(
                X, y, test_size=0.2, random_state=42
            )

            self.scaler = StandardScaler()
            X_train_scaled = self.scaler.fit_transform(X_train)

            self.model = RandomForestRegressor(
                n_estimators=50,  # Reduced for faster predictions
                max_depth=8,  # Reduced for faster predictions
                random_state=42,
                n_jobs=1,  # Single thread for consistency
            )
            self.model.fit(X_train_scaled, y_train)

            joblib.dump(self.model, self.model_path)
            joblib.dump(self.scaler, self.scaler_path)

            X_test_scaled = self.scaler.transform(X_test)
            score = self.model.score(X_test_scaled, y_test)
            print(f"Model R² score: {score:.4f}")

    def predict(self, features):
        """Make prediction for house price"""
        features_array = np.array(features).reshape(1, -1)
        features_scaled = self.scaler.transform(features_array)
        prediction = self.model.predict(features_scaled)[0]
        return prediction * 100000

    def get_feature_info(self):
        """Get information about the features"""
        return {
            "feature_names": list(self.feature_names),
            "num_features": len(self.feature_names),
            "description": "California housing dataset features",
        }

# Initialize model as singleton
ml_model = MLModel()

This page pydantic_models.py The file describes the Pydantic models to request and respond to verification data and stability.

app/pydantic_models.py:

from typing import List

from pydantic import BaseModel, Field

class PredictionRequest(BaseModel):
    features: List[float] = Field(
        ...,
        description="List of 8 features: MedInc, HouseAge, AveRooms, AveBedrms, Population, AveOccup, Latitude, Longitude",
        min_length=8,
        max_length=8,
    )

    model_config = {
        "json_schema_extra": {
            "examples": [
                {"features": [8.3252, 41.0, 6.984, 1.024, 322.0, 2.556, 37.88, -122.23]}
            ]
        }
    }

app/main.py: This file is a fastapi app, which describes EDPOINTS API.

import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from fastapi.responses import ORJSONResponse

from .ml_model import ml_model
from .pydantic_models import (
    PredictionRequest,
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Pre-load the model
    _ = ml_model.get_feature_info()
    yield

app = FastAPI(
    title="California Housing Price Prediction API",
    version="1.0.0",
    description="API for predicting California housing prices using Random Forest model",
    lifespan=lifespan,
    default_response_class=ORJSONResponse,
)

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "message": "Service is operational"}

@app.get("/model-info")
async def model_info():
    """Get information about the ML model"""
    try:
        feature_info = await asyncio.to_thread(ml_model.get_feature_info)
        return {
            "model_type": "Random Forest Regressor",
            "dataset": "California Housing Dataset",
            "features": feature_info,
        }
    except Exception:
        raise HTTPException(
            status_code=500, detail="Error retrieving model information"
        )

@app.post("/predict")
async def predict(request: PredictionRequest):
    """Make house price prediction"""
    if len(request.features) != 8:
        raise HTTPException(
            status_code=400,
            detail=f"Expected 8 features, got {len(request.features)}",
        )
    try:
        prediction = ml_model.predict(request.features)
        return {
            "prediction": float(prediction),
            "status": "success",
            "features_used": request.features,
        }
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception:
        raise HTTPException(status_code=500, detail="Prediction error")

Key Points:

lifespan Manager: Ensures that the ML model is loaded on time for app launch.
asyncio.to_thread: This is important because the Slikit-Read CPU CAPU (synced). Functional rope prevents the prevention of the Asynchli's Asynchronous event block, which allows the server to handle other applications at the same time.

EDPOINTS:

/health: The simple health check.
/model-info: Provides Metadata in ML model.
/predict: Accepting the list of items and returns the price of the house.

run_server.py: Contains text used to use the Fastapi app using Uvicorn.

import uvicorn

if __name__ == "__main__":

    uvicorn.run("app.main:app", host="localhost", port=8000, workers=4)

All files and configurations available in GitIb repository: KINGZPRI / oppression – Test-Fastapi

Obvious 3. Writing a checkpoint of the locust pressure

Now, let's build the scraper to check presses using the locusts.

tests/locustfile.py: This file describes the performance of characters.

import json
import logging
import random

from locust import HttpUser, task

# Reduce logging to improve performance
logging.getLogger("urllib3").setLevel(logging.WARNING)

class HousingAPIUser(HttpUser):
    def generate_random_features(self):
        """Generate random but realistic California housing features"""
        return [
            round(random.uniform(0.5, 15.0), 4),  # MedInc
            round(random.uniform(1.0, 52.0), 1),  # HouseAge
            round(random.uniform(2.0, 10.0), 2),  # AveRooms
            round(random.uniform(0.5, 2.0), 2),  # AveBedrms
            round(random.uniform(3.0, 35000.0), 0),  # Population
            round(random.uniform(1.0, 10.0), 2),  # AveOccup
            round(random.uniform(32.0, 42.0), 2),  # Latitude
            round(random.uniform(-124.0, -114.0), 2),  # Longitude
        ]

    @task(1)
    def model_info(self):
        """Test health endpoint"""
        with self.client.get("/model-info", catch_response=True) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Model info failed: {response.status_code}")

    @task(3)
    def single_prediction(self):
        """Test single prediction endpoint"""
        features = self.generate_random_features()


        with self.client.post(
            "/predict", json={"features": features}, catch_response=True, timeout=10
        ) as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    if "prediction" in data:
                        response.success()
                    else:
                        response.failure("Invalid response format")
                except json.JSONDecodeError:
                    response.failure("Failed to parse JSON")
            elif response.status_code == 503:
                response.failure("Service unavailable")
            else:
                response.failure(f"Status code: {response.status_code}")

Key Points:

Each user will wait between 0.5 and 2 seconds between the use tasks.
Creates random data for information data for predicting.
Each user will apply for a single_check application and 3 SINGLE_REDICTION requests.

Obvious 4. Running the pressure test

To check the performance of your app under load, start by starting your asynchronous schedule for one terminal.

Model loaded successfully
INFO:     Started server process [26216]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on  (Press CTRL+C to quit)

Open your browser and navigate to use active API documents to test your ENDPOINTS and make sure they work well.

Open a new new window, activate the visual nature, and navigate to the root indicator of your project to use the locusts of UI:

locust -f tests/locustfile.py --host

Reach the UI Portable UI In your browser.

UI Loan Loan Ui, set the total number of users to 500, the number of 10 users per second, and spend a minute.

During the test, locusts will show real-time statistics, including the amount of applications, failure and each response occasions.

When the test is complete, click on the chart tab view the graphs interact with the number of users, POUNDER SECOND, and answering times.

To conduct a logust without a web site and default the HTML message, use the following command:

locust -f tests/locustfile.py --host  --users 500 --spawn-rate 10 --run-time 60s --headless  --html report.html

After completing the examination, the HTML report named.html will be saved from your project director for later update.

Obvious The last thoughts

Our app can handle a large number of users as we use a simple machine learning model. The results indicate that the end of the model-incident has a great time to respond rather, impressive. This is a very good condition to check your app in your area before pressing it.

If you would like to hear this in the tape, please visit the KingzpPro / Stress-testing-surapy postportory and follow the instructions in the Scriptures.

Abid Awa (@ 1abidaswan) is a certified scientist for a scientist who likes the machine reading models. Currently, focus on the creation of the content and writing technical blogs in a machine learning and data scientific technology. Avid holds a Master degree in technical management and Bachelor degree in Telecommunication Engineering. His viewpoint builds AI product uses a Graph Neural network for students who strive to be ill.

Source link

nimda August 15, 2025

0 5 6 minutes read