Vengala Vinay

Having 9+ Years of Experience in Software Development

Home » Fixing Slow Largest Contentful Paint (LCP) caused by unoptimized database queries in Legacy Python Codebases Without Breaking API Contracts

Fixing Slow Largest Contentful Paint (LCP) caused by unoptimized database queries in Legacy Python Codebases Without Breaking API Contracts

Identifying the Database Bottleneck: A Deep Dive into Slow LCP

Largest Contentful Paint (LCP) is a critical user experience metric, directly impacting perceived page load speed and, consequently, conversion rates and SEO rankings. When LCP suffers, especially in legacy Python applications, the root cause often lies within unoptimized database queries that are executed synchronously during the initial page render. These queries, frequently triggered by template rendering or API endpoint logic that hasn’t been refactored for asynchronous operations, can block the main thread, delaying the rendering of the primary content element. Our first step is to pinpoint these specific queries.

We’ll leverage a combination of application-level profiling and database-level analysis. For Python applications, tools like cProfile or third-party APM (Application Performance Monitoring) solutions are invaluable. However, for a targeted approach on legacy systems where integrating new tools might be complex, we can start with manual logging and direct database query analysis.

Manual Profiling with Logging

Instrumenting your Python code to log the duration of database calls is a straightforward yet effective method. This involves wrapping your ORM (Object-Relational Mapper) calls or raw SQL executions with timing logic. Consider a common scenario where data is fetched for a Django or Flask template.

Let’s assume a hypothetical view function in a Flask application:

import time
from flask import Flask, render_template
from my_models import get_user_profile, get_recent_orders

app = Flask(__name__)

@app.route('/user/')
def user_profile_page(user_id):
    start_time = time.time()
    profile_data = get_user_profile(user_id)
    profile_fetch_time = time.time() - start_time
    print(f"DEBUG: get_user_profile for {user_id} took {profile_fetch_time:.4f} seconds")

    start_time = time.time()
    recent_orders = get_recent_orders(user_id, limit=10)
    orders_fetch_time = time.time() - start_time
    print(f"DEBUG: get_recent_orders for {user_id} took {orders_fetch_time:.4f} seconds")

    total_db_time = profile_fetch_time + orders_fetch_time
    print(f"DEBUG: Total DB time for user_profile_page for {user_id}: {total_db_time:.4f} seconds")

    return render_template('profile.html', profile=profile_data, orders=recent_orders)

# Assume get_user_profile and get_recent_orders are functions that interact with the DB
# e.g., using SQLAlchemy or Django ORM

By observing these log messages under load, we can identify which database calls are consuming the most time. If profile_fetch_time or orders_fetch_time consistently exceeds a few hundred milliseconds, it’s a strong indicator of a database bottleneck.

Database-Level Query Analysis

Once a slow query is suspected, we need to examine its execution plan. Most relational databases provide tools for this. For PostgreSQL, we use EXPLAIN ANALYZE. For MySQL, it’s EXPLAIN.

Let’s assume get_user_profile executes a query similar to this:

SELECT id, username, email, bio, avatar_url
FROM users
WHERE id = 'some-user-id';

If this query is slow, running EXPLAIN ANALYZE on it (replacing the placeholder with an actual ID) will reveal if an index is missing or if the database is performing a full table scan. For example, in PostgreSQL:

EXPLAIN ANALYZE SELECT id, username, email, bio, avatar_url
FROM users
WHERE id = 'a1b2c3d4-e5f6-7890-1234-567890abcdef';

A typical slow output might show a “Seq Scan” (Sequential Scan) on a large table, indicating a lack of an appropriate index on the id column. The output would look something like this (simplified):

                                                     QUERY PLAN
-----------------------------------------------------------------------------------------------------
 Seq Scan on users  (cost=0.00..15000.00 rows=1000000 width=100) (actual time=0.050..120.500 rows=1000000 loops=1)
   Filter: (id = 'a1b2c3d4-e5f6-7890-1234-567890abcdef'::uuid)
   Rows Removed by Filter: 999999
 Planning Time: 0.100 ms
 Execution Time: 130.600 ms

The high Execution Time and the presence of Seq Scan are clear red flags. The fix here would be to add an index:

CREATE INDEX idx_users_id ON users (id);

Similarly, for the get_recent_orders query, if it’s something like:

SELECT order_id, order_date, total_amount
FROM orders
WHERE user_id = 'some-user-id'
ORDER BY order_date DESC
LIMIT 10;

An EXPLAIN ANALYZE might reveal a scan on the orders table and a sort operation if there isn’t a composite index. The solution would be:

CREATE INDEX idx_orders_user_id_date ON orders (user_id, order_date DESC);

Refactoring for Asynchronous Data Fetching and Caching

Once the slow queries are identified and optimized at the database level (e.g., by adding indexes), the next crucial step is to refactor the application code to avoid blocking the main thread during LCP-critical rendering paths. This involves moving synchronous database calls to asynchronous operations or implementing effective caching strategies. The goal is to ensure that the initial HTML document is sent to the browser as quickly as possible, even if some data is loaded progressively.

Leveraging Asynchronous Operations (If Supported)

Modern Python web frameworks and ORMs increasingly support asynchronous operations. If your legacy codebase uses an ORM like SQLAlchemy with its async capabilities or a framework like FastAPI, you can rewrite synchronous database calls to be non-blocking. For older Django or Flask applications, this might require significant refactoring or the introduction of background task queues.

Consider an asynchronous version of the Flask example. This requires an ASGI server (like Uvicorn) and an async-compatible database driver and ORM setup. This is a substantial change for a legacy app, so it’s often more practical to use caching or defer non-critical data loading.

import time
from flask import Flask, render_template
from my_async_models import get_user_profile_async, get_recent_orders_async # Hypothetical async functions

app = Flask(__name__)

@app.route('/user/')
async def user_profile_page_async(user_id): # Note the 'async' keyword
    # In a real async scenario, these would be awaited
    # For demonstration, we'll simulate async work with sleep
    profile_data = await get_user_profile_async(user_id)
    recent_orders = await get_recent_orders_async(user_id, limit=10)

    # The rendering itself is still synchronous here, but data fetching is non-blocking
    return render_template('profile.html', profile=profile_data, orders=recent_orders)

# To run this, you'd typically use:
# uvicorn main:app --reload --workers 1

However, for many legacy applications, a full async rewrite is not feasible. In such cases, focusing on caching and deferring non-essential data is more pragmatic.

Implementing Caching Strategies

Caching is often the most effective way to reduce database load and improve LCP without extensive code rewrites. We can cache query results in memory (e.g., using `functools.lru_cache` for simple cases) or, more robustly, using external caching systems like Redis or Memcached.

Let’s refactor the original synchronous Flask example to use Redis for caching user profile data. We’ll need a Redis client library (e.g., redis-py).

import time
import json
import redis
from flask import Flask, render_template
from my_models import get_user_profile, get_recent_orders # Original synchronous functions

app = Flask(__name__)

# Configure Redis connection
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)
CACHE_EXPIRY_SECONDS = 300 # Cache for 5 minutes

def get_cached_user_profile(user_id):
    cache_key = f"user_profile:{user_id}"
    cached_data = redis_client.get(cache_key)
    if cached_data:
        print(f"DEBUG: Cache hit for user_profile:{user_id}")
        return json.loads(cached_data)
    else:
        print(f"DEBUG: Cache miss for user_profile:{user_id}")
        profile_data = get_user_profile(user_id) # Original DB call
        if profile_data:
            # Serialize and store in cache
            redis_client.setex(cache_key, CACHE_EXPIRY_SECONDS, json.dumps(profile_data))
        return profile_data

@app.route('/user/')
def user_profile_page_cached(user_id):
    start_time = time.time()
    profile_data = get_cached_user_profile(user_id) # Use cached version
    profile_fetch_time = time.time() - start_time
    print(f"DEBUG: Profile data retrieval (incl. cache check) took {profile_fetch_time:.4f} seconds")

    # Orders might still be fetched directly if they change frequently or are less critical for LCP
    # Or they could also be cached with a different expiry.
    start_time = time.time()
    recent_orders = get_recent_orders(user_id, limit=10)
    orders_fetch_time = time.time() - start_time
    print(f"DEBUG: get_recent_orders for {user_id} took {orders_fetch_time:.4f} seconds")

    total_db_time = profile_fetch_time + orders_fetch_time # Note: profile_fetch_time now includes cache lookup
    print(f"DEBUG: Total processing time for user_profile_page_cached for {user_id}: {total_db_time:.4f} seconds")

    return render_template('profile.html', profile=profile_data, orders=recent_orders)

# Ensure Redis server is running and accessible at localhost:6379
# Install redis-py: pip install redis

In this cached version, the get_user_profile call is wrapped. If the data is in Redis, it’s returned immediately, significantly reducing latency and database load. If not, the original query runs, and its result is then cached for subsequent requests. The key is to identify data that is frequently read but infrequently written, making it an ideal candidate for caching.

Deferring Non-Critical Data Loading

Not all data displayed on a page is essential for the initial render and LCP. Elements like user comments, related articles, or secondary user information might be loaded after the main content has rendered. This can be achieved using JavaScript to fetch data via AJAX calls once the page is interactive.

Consider the profile.html template. If the ‘recent_orders’ section is not critical for the initial LCP, we can modify the template and view:

# In your Flask view (user_profile_page_cached)
@app.route('/user/')
def user_profile_page_deferred(user_id):
    profile_data = get_cached_user_profile(user_id) # Still cached for speed
    # Do NOT fetch recent_orders here for LCP

    # Pass only essential data to the initial render
    return render_template('profile_deferred.html', profile=profile_data, user_id=user_id)

# Create a new endpoint for fetching orders
@app.route('/api/user//orders')
def get_user_orders_api(user_id):
    recent_orders = get_recent_orders(user_id, limit=10) # Fetch directly when requested
    return json.dumps(recent_orders)

And the corresponding template (profile_deferred.html):

<!DOCTYPE html>
<html>
<head>
    <title>User Profile</title>
    <!-- Other head elements -->
</head>
<body>
    <!-- Main content that determines LCP -->
    <div class="profile-main">
        <h1>{{ profile.username }}</h1>
        <img src="{{ profile.avatar_url }}" alt="Avatar">
        <p>{{ profile.bio }}</p>
    </div>

    <!-- Placeholder for orders, to be loaded by JavaScript -->
    <div id="orders-section">
        Loading orders...
    </div>

    <script>
        document.addEventListener('DOMContentLoaded', function() {
            const userId = "{{ user_id }}"; // Get user ID from template
            fetch(`/api/user/${userId}/orders`)
                .then(response => response.json())
                .then(orders => {
                    const ordersSection = document.getElementById('orders-section');
                    if (orders.length > 0) {
                        let ordersHtml = '<h2>Recent Orders</h2><ul>';
                        orders.forEach(order => {
                            ordersHtml += `<li>${order.order_date}: ${order.total_amount}</li>`;
                        });
                        ordersHtml += '</ul>';
                        ordersSection.innerHTML = ordersHtml;
                    } else {
                        ordersSection.innerHTML = '<p>No recent orders found.</p>';
                    }
                })
                .catch(error => {
                    console.error('Error fetching orders:', error);
                    document.getElementById('orders-section').innerHTML = '<p>Could not load orders.</p>';
                });
        });
    </script>
</body>
</html>

By deferring the fetching and rendering of the ‘orders’ section, the initial HTML document can be sent much faster, improving LCP. The JavaScript then asynchronously fetches and displays the order data once the main content is visible.

Maintaining API Contracts and Testing

When refactoring legacy code, especially for performance improvements, it’s paramount to ensure that existing API contracts are not broken. For internal APIs or backend-to-frontend communication, this means the structure and content of the data returned should remain consistent, even if the underlying implementation changes.

Contract Testing

Implement contract tests to verify that your API endpoints (both external and internal) return data in the expected format. Tools like Pact can be used for consumer-driven contract testing, ensuring that the consumer (e.g., a frontend JavaScript application) and the provider (your Python backend) agree on the API’s structure.

For simpler cases, you can write integration tests that specifically assert the JSON structure and data types of API responses. If you’re using Django REST Framework or Flask-RESTful, these frameworks often have built-in testing utilities that make this easier.

# Example using Flask's test_client
from my_app import app # Assuming your Flask app instance is named 'app'
import json

def test_user_profile_api_contract():
    client = app.test_client()
    user_id = 'test-user-123' # Use a known test user ID

    # First, ensure the initial page render endpoint returns expected structure
    response_page = client.get(f'/user/{user_id}')
    assert response_page.status_code == 200
    # Basic check for presence of main profile elements in HTML
    assert b'<h1>Test User</h1>' in response_page.data # Example assertion

    # Test the API endpoint for deferred data
    response_api = client.get(f'/api/user/{user_id}/orders')
    assert response_api.status_code == 200
    data = json.loads(response_api.get_data(as_text=True))

    # Assertions on the contract of the orders API response
    assert isinstance(data, list)
    if data: # If there are orders, check the structure of the first one
        order = data[0]
        assert 'order_id' in order
        assert 'order_date' in order
        assert 'total_amount' in order
        assert isinstance(order['order_id'], str) # Or int, depending on your schema
        assert isinstance(order['total_amount'], (int, float))
    print("API contract test passed for user orders.")

# Run this test using pytest or your preferred test runner
# pytest your_test_file.py

Gradual Rollout and Monitoring

Introduce changes incrementally. Deploy caching or deferred loading for a subset of users or traffic first. Continuously monitor LCP and other performance metrics, as well as error rates, using your APM tools and real user monitoring (RUM) services. This allows you to catch regressions or unexpected side effects early.

Key metrics to watch:

  • Largest Contentful Paint (LCP)
  • First Contentful Paint (FCP)
  • Time to Interactive (TTI)
  • Server Response Time (TTFB)
  • Database query latency
  • Cache hit/miss ratio
  • Application error rates

By combining meticulous profiling, strategic refactoring (caching, deferral), and robust testing, you can effectively address slow LCP issues stemming from database bottlenecks in legacy Python codebases without compromising existing API contracts.