[Python] My endpoints through locust are running very slowly

I use FastAPI, MySQL and asynchronous approach. The problem is that with a heavy load, for example, if there are 100 users through locust, I make a ramp for 5 users, then requests can take 20-40 seconds to complete.

I tried to make connections with the database for each core, it only made it worse. I am running the application and testing on my machine and on a remote server and the problem was both there and there.

Who has encountered such a thing, how did you solve the problem? GPT didn't help me. I will provide the code for my database class and the endpoint that is slowing down (I have also set up indexes). (the server and the machine have 4 cores and 16GB ram And it's 60% loaded.)

Base = declarative_base()

class Database:
def __init__(self):
self.test_mode = os.getenv("TEST_MODE", "false").lower() == "true"
self.db_name = "test" if self.test_mode else os.getenv('DB_NAME')
# URL for connecting to an existing database
self.SQLALCHEMY_DATABASE_URL = (
f"mysql+aiomysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@"
f"{os.getenv('DB_HOST')}:3306/{self.db_name}?charset=utf8mb4"
)
logger.info(f"DATABASE_URL {self.SQLALCHEMY_DATABASE_URL}")
# Creating a single asynchronous engine
self.engine = create_async_engine(
self.SQLALCHEMY_DATABASE_URL,
**database_settings.DB_CONFIG,
execution_options={"isolation_level": "READ COMMITTED"} # Important for high-load systems
)

# Creating a session factory
self.session_factory = sessionmaker(
autocommit=False,
autoflush=False,
bind=self.engine,
class_=AsyncSession,
expire_on_commit=False
)

async def initialize_database(self):
"""
Initializing the database and creating tables
"""
try:
# Create a database if it does not exist
init_engine = create_async_engine(
f"mysql+aiomysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@"
f"{os.getenv('DB_HOST')}:3306?charset=utf8mb4",
**{k: v for k, v in database_settings.DB_CONFIG.items() if k != 'pool_size'}
)

async with init_engine.begin() as conn:
await conn.execute(text(f"CREATE DATABASE IF NOT EXISTS {self.db_name} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"))
await conn.execute(text(f"SET SESSION sql_mode = (SELECT REPLACE(@@SESSION.sql_mode, 'ONLY_FULL_GROUP_BY', ''));"))

logger.info("DATABASE SCHEMA CREATED OR EXISTS")

# Creating tables using the engine
async with self.engine.begin() as connection:
def sync_inspect(conn):
inspector = inspect(conn)
return inspector.get_table_names()

existing_tables = await connection.run_sync(sync_inspect)
logger.info(f"Existing tables: {existing_tables}")

if not existing_tables:
await connection.run_sync(Base.metadata.create_all)
logger.info("Tables created successfully")
else:
logger.info("Tables already exist")

# Checking the availability of all necessary tables
required_tables = {
'users', 'auth', 'history', 'websites', 'user_files', 'messages',
'assistants', 'assistant_usage_history', 'assistant_tags',
'messages_websites', 'tags', 'user_assistant_settings'
}
missing_tables = required_tables - set(existing_tables)

if missing_tables:
logger.info(f"Creating missing tables: {missing_tables}")
await connection.run_sync(Base.metadata.create_all)

# Connection test
async with self.engine.connect() as conn:
result = await conn.execute(text("SELECT 1"))
result.fetchone()

except Exception as e:
logger.error(f"Database initialization error: {str(e)}")
raise

@contextlib.asynccontextmanager
async def get_db_session(self):
"""
Context manager for working with a database session.
Automatically selects a connection based on the current thread.
This ensures that each thread uses its own dedicated connection,
preventing "Packet sequence number wrong" errors.
"""
session = self.session_factory()
try:
yield session
await session.commit()
except HTTPException:
raise
except Exception as e:
await session.rollback()
logger.error(f"Database error occurred: {str(e)}")
if "Packet sequence number wrong" in str(e):
logger.critical("Detected packet sequence error. Consider recycling the connection pool.")
raise HTTPException(
status_code=500,
detail="Database error occurred"
)
finally:
await session.close()


@router.post(
"/create_share_link/{history_url}", response_model=schemas.ShareUrlResponse
)
async def share_history(
history_url: str,
request: Request,
user_id: int = Depends(get_current_user),
session: AsyncSession = Depends(get_db),
) -> schemas.ShareUrlResponse:
"""
Opening access to the chat history
"""
try:
if not re.match(r"^[a-z0-9\-]{1,35}\-[a-f0-9]{22}$", history_url):
raise HTTPException(status_code=400, detail="Invalid history URL format")

# Checking the ownership of the history
history_query = (
select(History)
.join(Messages, History.id == Messages.history_id)
.where(and_(History.url == history_url, Messages.user_id == user_id))
.limit(1)
)

history_result = await session.execute(history_query)
history = history_result.scalar_one_or_none()

if not history:
raise HTTPException(
status_code=404,
detail="The history was not found or does not belong to you",
)

if not history.is_shared:
history.is_shared = True
await session.commit()

# Creating a link for public access
share_url = f"/history/{history_url}"

return schemas.ShareUrlResponse(share_url=share_url)

except SQLAlchemyError as e:
logger.error(f"Database error in share_history: {str(e)}")
await session.rollback()
raise HTTPException(status_code=500, detail="Database error occurred")
except HTTPException:
# Re-raise HTTP exceptions
raise
except Exception as e:
logger.error(f"Unexpected error in share_history: {str(e)}")
await session.rollback()
raise HTTPException(status_code=500, detail="An unexpected error occurred")


async def get_db():
async with database.get_db_session() as session:
try:
yield session
except HTTPException as e:
raise e
except Exception as e:
logger.error(f"Database session error: {e}")
raise

DB_CONFIG: dict = {
"pool_size": 20,
"max_overflow": 30,
"pool_timeout": 10,
"pool_recycle": 1800,
"pool_pre_ping": True,
"echo": False,
}

Continue reading...