Réglage des performances pour les applications mssql-python

Le pilote mssql-python offre plusieurs fonctionnalités et schémas pour optimiser les performances des applications SQL Server, notamment le pooling de connexions, l’optimisation des requêtes et les opérations en masse.

Gestion des connexions

Utiliser le regroupement de connexions

La gestion des connexions est intégrée nativement. Lorsque vous appelez conn.close(), la connexion retourne au pool pour être réutilisée au lieu d’être détruite, donc les appels suivants connect() sautent la poignée de main coûteuse :

import mssql_python

def get_data():
    conn = mssql_python.connect(
        "Server=<server>.database.windows.net;Database=<database>;"
        "Authentication=ActiveDirectoryDefault;Encrypt=yes"
    )
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
        return cursor.fetchall()
    finally:
        conn.close()

Configurez la taille du pool pour la charge de travail

Ajustez la taille du pool en fonction de vos besoins de concurrence. Si votre application gère de nombreux utilisateurs simultanés, augmentez le pool. Pour des charges de travail plus légères, un pool plus petit économise les ressources du serveur :

import mssql_python

mssql_python.pooling(
    max_size=50,      # Default is 100; reduce or increase for your workload
    idle_timeout=600  # Seconds before idle connections are recycled
)

Réutiliser les connexions au sein des opérations

Ouvrir une nouvelle connexion pour chaque requête ajoute de la surcharge même avec le pooling. Au lieu de cela, maintenez une seule connexion pendant la durée d’une opération logique :

# Bad: New connection per query
def bad_pattern(product_ids):
    for pid in product_ids:
        conn = mssql_python.connect(connection_string)
        cursor = conn.cursor()
        cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE ProductID = %(pid)s", {"pid": pid})
        row = cursor.fetchone()
        process(row)
        conn.close()

# Good: Single connection for all queries
def good_pattern(product_ids):
    conn = mssql_python.connect(connection_string)
    cursor = conn.cursor()
    try:
        for pid in product_ids:
            cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE ProductID = %(pid)s", {"pid": pid})
            row = cursor.fetchone()
            process(row)
    finally:
        conn.close()

Maintenir les connexions ouvertes dans les services de longue durée

Les serveurs web, les travailleurs en file d’attente et les tâches planifiées qui s’exécutent en continu devraient maintenir les connexions ouvertes plutôt que de se connecter et se déconnecter à chaque opération. L’ouverture d’une connexion implique une poignée de main TCP, une négociation TLS et une authentification, ce qui peut prendre entre 50 et 200 ms selon la distance réseau et la méthode d’authentification. Pour un travailleur de file d’attente qui traite des milliers de messages par heure, cette surcharge s’accumule rapidement.

Gardez la connexion ouverte pendant toute la durée de vie de l’ouvrier et reconnectez-vous lorsque la connexion tombe en panne. Attendez entre les itérations pour éviter de surcharger le serveur lorsque la file d’attente est vide :

import mssql_python
import time

def run_worker(connection_string: str, poll_interval: float = 1.0):
    conn = None
    try:
        while True:
            try:
                if conn is None:
                    conn = mssql_python.connect(connection_string)
                cursor = conn.cursor()
                cursor.execute("SELECT TOP 1 * FROM dbo.JobQueue WHERE Status = 'Pending' ORDER BY CreatedDate")
                job = cursor.fetchone()
                if job:
                    try:
                        process_job(job)
                        cursor.execute("UPDATE dbo.JobQueue SET Status = 'Done' WHERE JobID = %(job_id)s", {"job_id": job[0]})
                    except Exception:
                        cursor.execute("UPDATE dbo.JobQueue SET Status = 'Failed' WHERE JobID = %(job_id)s", {"job_id": job[0]})
                    conn.commit()
                else:
                    time.sleep(poll_interval)  # No work available, wait before polling again
            except mssql_python.OperationalError:
                # Connection lost, reconnect on next iteration
                conn = None
                time.sleep(poll_interval)
    finally:
        if conn is not None:
            conn.close()

Avec le pooling de connexion activé (par défaut), le pool gère les connexions inactives pour vous. Mais si vous désactivez la mise en pool ou utilisez une seule connexion dédiée, définissez Connection Timeout et Command Timeout dans votre chaîne de connexion afin de détecter rapidement les connexions obsolètes au lieu de rester bloqué.

Optimisation des requêtes

Fetch n’avait besoin que de données

Sélectionner uniquement les colonnes utilisées par votre application réduit le transfert réseau, la consommation de mémoire et le temps d’exécution des requêtes.

# Bad: Select all columns
cursor.execute("SELECT * FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": 1})

# Good: Select specific columns
cursor.execute("""
    SELECT SalesOrderID, OrderDate, TotalDue
    FROM Sales.SalesOrderHeader
    WHERE CustomerID = %(customer_id)s
""", {"customer_id": 1})

Utiliser des méthodes de récupération appropriées

Le pilote propose plusieurs méthodes de récupération. Utilisez celui qui correspond à la taille de votre résultat :

• fetchval() renvoie une seule valeur scalaire avec un coût minimal.
• fetchall() charge l’ensemble des résultats en mémoire, ce qui fonctionne bien pour les petites tables.
• fetchmany(n) récupère les lignes par lots, en maintenant une consommation mémoire constante pour de grands ensembles de résultats.

La taille de lot appropriée fetchmany() dépend de la largeur des lignes. Pour les lignes étroites (quelques petites colonnes, environ 1 Ko chacune), 1 000 lignes maintiennent chaque lot autour de 1 Mo de mémoire. Pour les rangées plus larges avec de grandes chaînes ou colonnes binaires, utilisez une taille de lot plus petite. Commencez par 1 000 et ajustez en fonction de vos données.

def process_batch(rows):
    # Example: print each row. Replace with your own logic.
    for row in rows:
        print(row)

# Single value
cursor.execute("SELECT COUNT(*) FROM Production.Product")
count = cursor.fetchval()

# Small result set
cursor.execute("SELECT ProductCategoryID, Name FROM Production.ProductCategory")
categories = cursor.fetchall()

# Large result set - process in batches
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader")
while True:
    batch = cursor.fetchmany(1000)
    if not batch:
        break
    process_batch(batch)

Utiliser la pagination côté serveur

Au lieu de récupérer toutes les lignes et de découper en Python, utilisez OFFSET/FETCH NEXT pour récupérer uniquement la page dont vous avez besoin.

def get_page(cursor, page: int, page_size: int = 50) -> list:
    """Get paginated results efficiently."""
    offset = (page - 1) * page_size

    cursor.execute("""
        SELECT ProductID, Name, ListPrice
        FROM Production.Product
        ORDER BY ProductID
        OFFSET %(offset)s ROWS
        FETCH NEXT %(page_size)s ROWS ONLY
    """, {"offset": offset, "page_size": page_size})

    return cursor.fetchall()

Utilisez SET NOCOUNT ON

Par défaut, SQL Server envoie un message « lignes affectées » après chaque instruction DML. SET NOCOUNT ON supprime ces messages et réduit le trafic réseau. C’est un réglage au niveau de la session, donc il faut le définir une fois après la connexion plutôt que de l’intégrer à chaque requête.

# Set once after connecting
cursor.execute("SET NOCOUNT ON")

# All subsequent statements on this connection skip the row-count message
cursor.execute(
    "INSERT INTO Log (Message) VALUES (%(message)s)",
    {"message": "Log entry"}
)

Choisissez la bonne méthode d’insertion

Le pilote propose trois méthodes d’insertion de données, chacune adaptée à une échelle différente :

Méthode	Nombre de lignes	Pourquoi
execute()	1 ligne par appel	À utiliser pour des opérations sur une seule ligne, comme les soumissions de formulaires ou les gestionnaires d’API, lorsque vous avez besoin immédiatement de l’ID inséré.
executemany()	~10-1 000 lignes	Utiliser la liaison de paramètres par colonne pour de meilleures performances qu’un traitement en boucle. Envoie chaque ligne comme une instruction paramétrée.
bulkcopy()	Des centaines de rangées et plus	Utilise le protocole TDS d’insertion en bloc, nettement plus efficace que les insertions ligne par ligne. Idéal pour les charges de données, les migrations et le traitement par lots.

Pour plus de détails et d’exemples, voir Chargement et schémas de déplacement des données.

Inserts simples avec exécute()

À utiliser pour des insertions ponctuelles lorsque vous avez besoin d’un résultat immédiat. Production.Product possède plusieurs colonnes NOT NULL sans valeurs par défaut, donc l’insert les liste toutes :

from datetime import datetime

cursor.execute(
    """
    INSERT INTO Production.Product
        (Name, ProductNumber, SafetyStockLevel, ReorderPoint,
         StandardCost, ListPrice, DaysToManufacture, SellStartDate)
    VALUES (%(name)s, %(number)s, %(safety)s, %(reorder)s,
            %(cost)s, %(price)s, %(days)s, %(start)s)
    """,
    {
        "name": "Widget", "number": "WG-1001",
        "safety": 100, "reorder": 75,
        "cost": 12.50, "price": 19.99,
        "days": 1, "start": datetime(2024, 1, 1),
    },
)
conn.commit()

Insertions par lot avec executemany()

executemany() lie les paramètres colonnes par colonnes et les envoie efficacement. Utilisez-le pour des lots modérés au lieu d’appeler execute() en boucle. Notez que executemany() nécessite des marqueurs ? positionnels avec une liste de tuples, tandis que execute() prend en charge à la fois ? et des paramètres nommés %(name)s avec des dictionnaires. Voir requêtes paramétrées pour les détails sur chaque style.

rows = [
    ("Widget A", "WG-1001", 100, 75, 12.50, 19.99, 1, datetime(2024, 1, 1)),
    ("Widget B", "WG-1002", 100, 75, 15.00, 24.99, 1, datetime(2024, 1, 1)),
    ("Widget C", "WG-1003", 100, 75, 18.00, 29.99, 1, datetime(2024, 1, 1)),
]

cursor.executemany(
    """
    INSERT INTO Production.Product
        (Name, ProductNumber, SafetyStockLevel, ReorderPoint,
         StandardCost, ListPrice, DaysToManufacture, SellStartDate)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """,
    rows,
)
conn.commit()

Copie en masse pour les charges volumineuses

Lorsque le débit compte plus que le contrôle par ligne, on passe à bulkcopy(). Il transmet les lignes via le protocole d’insertion en masse TDS et évite la surcharge associée à chaque ligne des requêtes paramétrées. Le point de bascule exact à partir duquel bulkcopy() surpasse executemany() dépend de la largeur des lignes et de la latence du réseau, mais il se situe généralement de l’ordre de quelques centaines de lignes. Pour de très petits lots, executemany() c’est plus simple car bulkcopy() cela crée une connexion interne séparée et fait des commits automatiques.

Contrairement à execute() et executemany(), bulkcopy() associe les valeurs aux colonnes par position, et non à l’aide d’une liste de colonnes INSERT. Passez column_mappings pour nommer les colonnes de destination que vous chargez, afin que les tuples sources s’alignent avec les colonnes de droite au lieu de la colonne d’identité principale de la table :

result = cursor.bulkcopy(
    "Production.Product",
    rows,
    column_mappings=["Name", "ProductNumber", "SafetyStockLevel", "ReorderPoint",
                     "StandardCost", "ListPrice", "DaysToManufacture", "SellStartDate"],
)
print(f"Copied {result['rows_copied']} rows")

Pour des charges très importantes, utilisez un générateur pour éviter de charger l’ensemble de données en mémoire et configurez batch_size pour valider périodiquement :

import csv

def csv_rows(path):
    with open(path, newline="") as f:
        reader = csv.reader(f)
        next(reader)  # Skip header
        for row in reader:
            yield tuple(row)

cursor.bulkcopy(
    "Production.Product",
    csv_rows("products.csv"),
    column_mappings=["Name", "ProductNumber", "SafetyStockLevel", "ReorderPoint",
                     "StandardCost", "ListPrice", "DaysToManufacture", "SellStartDate"],
    batch_size=5000,
)

Stratégies de mise en cache

Pour les données de référence qui changent rarement (catégories, tables de recherche, configuration), mettez en cache les résultats dans votre application au lieu de rechercher chaque requête.

Le functools.lru_cache de Python fournit une mémoïsation simple, mais conserve les résultats en cache indéfiniment jusqu'au redémarrage du processus. Si les données sous-jacentes peuvent changer, utilisez cachetools.TTLCache pour rafraîchir automatiquement après une limite de temps :

from cachetools import TTLCache, cached

category_cache = TTLCache(maxsize=1, ttl=300)  # Refresh every 5 minutes

@cached(category_cache)
def get_categories(connection_string: str) -> tuple:
    conn = mssql_python.connect(connection_string)
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT ProductCategoryID, Name FROM Production.ProductCategory")
        return cursor.fetchall()
    finally:
        conn.close()

Optimisation réseau

Minimiser les allers-retours

Chaque requête est un aller-retour réseau vers le serveur. Combinez les requêtes apparentées en un seul lot et utilisez nextset() pour avancer dans les ensembles de résultats :

# Bad: Multiple round trips
cursor.execute("SELECT CustomerID, AccountNumber FROM Sales.Customer WHERE CustomerID = %(customer_id)s", {"customer_id": 1})
customer = cursor.fetchone()
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": 1})
orders = cursor.fetchall()
cursor.execute("SELECT COUNT(*) FROM Sales.SalesOrderDetail WHERE SalesOrderID IN (SELECT SalesOrderID FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s)", {"customer_id": 1})
detail_count = cursor.fetchval()

# Good: Single round trip
cursor.execute("""
    SELECT CustomerID, AccountNumber FROM Sales.Customer WHERE CustomerID = %(customer_id)s;
    SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s;
    SELECT COUNT(*) FROM Sales.SalesOrderDetail WHERE SalesOrderID IN (SELECT SalesOrderID FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s);
""", {"customer_id": 1})

customer = cursor.fetchone()
cursor.nextset()
orders = cursor.fetchall()
cursor.nextset()
detail_count = cursor.fetchval()

Utiliser le traitement côté serveur pour une logique complexe

Poussez l’agrégation et le filtrage vers SQL Server au lieu de récupérer des lignes brutes et de les traiter en Python. Le serveur retourne une seule ligne résumée au lieu de potentiellement des milliers de lignes de détails :

cursor.execute("""
    SELECT p.Name, COUNT(sod.SalesOrderDetailID) AS OrderCount, SUM(sod.LineTotal) AS TotalSales
    FROM Production.Product p
    JOIN Sales.SalesOrderDetail sod ON p.ProductID = sod.ProductID
    WHERE p.ProductID = %(product_id)s
    GROUP BY p.Name
""", {"product_id": 707})

Évitez les opérations de curseur entrelacé

Le pilote mssql-python ne prend pas en charge les ensembles de résultats actifs multiples (MARS). Un seul curseur peut avoir une requête active par connexion. Récupérez complètement le premier ensemble de résultats avant d’exécuter la requête suivante, ou utilisez une seconde connexion :

connection_string = (
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

# Option 1: Fetch first, then query (single connection)
# Warning: This is an N+1 pattern. Each iteration is a round trip.
# Use this only when the JOIN in Option 2 is not possible.
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("SELECT ProductID FROM Production.Product WHERE ProductSubcategoryID = 1")
product_ids = [row[0] for row in cursor.fetchall()]

for pid in product_ids:
    cursor.execute("SELECT ProductID, LocationID, Quantity FROM Production.ProductInventory WHERE ProductID = %(product_id)s", {"product_id": pid})
    inventory = cursor.fetchone()

conn.close()

# Option 2: Use a JOIN instead of N+1 queries (preferred)
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("""
    SELECT p.ProductID, p.Name, i.Quantity
    FROM Production.Product p
    LEFT JOIN Production.ProductInventory i ON p.ProductID = i.ProductID
    WHERE p.ProductSubcategoryID = 1
""")
results = cursor.fetchall()
conn.close()

Gestion de la mémoire

Traiter les gros résultats en blocs

Charger une table de plusieurs millions de lignes dans une liste consomme une mémoire proportionnelle à l’ensemble complet des résultats. Utilisez OFFSET et FETCH NEXT pour parcourir les données avec pagination côté serveur et traiter un bloc à la fois.

def quote_id(identifier: str) -> str:
    """Quote a possibly schema-qualified SQL identifier to prevent SQL injection."""
    return ".".join("[" + part.replace("]", "]]") + "]" for part in identifier.split("."))

def process_large_table(cursor, table: str, columns: list[str], key_column: str, processor, chunk_size: int = 10000):
    """Process large table without loading all data."""
    safe_table = quote_id(table)
    safe_key = quote_id(key_column)
    col_list = ", ".join(quote_id(c) for c in columns)
    cursor.execute(f"SELECT COUNT(*) FROM {safe_table}")
    total = cursor.fetchval()

    offset = 0
    while offset < total:
        cursor.execute(f"""
            SELECT {col_list} FROM {safe_table}
            ORDER BY {safe_key}
            OFFSET ? ROWS
            FETCH NEXT ? ROWS ONLY
        """, (offset, chunk_size))

        chunk = cursor.fetchall()
        processor(chunk)

        offset += chunk_size
        print(f"Processed {min(offset, total)}/{total}")

# key_column must be unique, otherwise rows can be duplicated or skipped across pages
process_large_table(
    cursor,
    "Production.TransactionHistory",
    ["TransactionID", "ProductID", "Quantity", "ActualCost"],
    "TransactionID",
    lambda chunk: None,  # replace with your row-processing logic
)

Utiliser les générateurs pour le streaming

Un générateur Python qui encapsule fetchmany() maintient une utilisation mémoire constante, quelle que soit la taille de la table. L’appelant itère rangée par ligne sans charger l’ensemble complet des résultats. Pour une source extra-grande, combinez les tables avec UNION ALL et diffusez le résultat combiné de la même manière.

def stream_query(cursor, query: str, params: dict = None, batch_size: int = 1000):
    cursor.execute(query, params or {})
    
    while True:
        batch = cursor.fetchmany(batch_size)
        if not batch:
            break
        for row in batch:
            yield row

# Union the live and archive transaction tables into one extra-large result set
query = """
    SELECT TransactionID, ProductID, Quantity, ActualCost FROM Production.TransactionHistory
    UNION ALL
    SELECT TransactionID, ProductID, Quantity, ActualCost FROM Production.TransactionHistoryArchive
"""

count = 0
for row in stream_query(cursor, query, batch_size=5000):
    count += 1
print(f"Streamed {count} rows")

Nettoyage rapide des ressources

Les connexions non fermées monopolisent les ressources du serveur et peuvent épuiser le pool de connexions. Utilisez un gestionnaire de contexte pour garantir le nettoyage même en cas d’exceptions.

from contextlib import contextmanager

@contextmanager
def database_connection(connection_string: str):
    conn = mssql_python.connect(connection_string)
    try:
        yield conn
    finally:
        conn.close()

with database_connection(connection_string) as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
    data = cursor.fetchall()

Surveiller l’utilisation de la mémoire

Les grands ensembles de résultats, les caches à longue durée de vie et les objets de connexion consomment tous de la mémoire. Si votre application s’exécute en tant que service, des fuites de mémoire provenant de curseurs non fermés ou de caches non bornés peuvent finalement entraîner la mort du processus par l’OS ou l’exécution du conteneur.

Utilisez le module tracemalloc de Python pour prendre des instantanés de la mémoire et trouver les plus grandes allocations.

import tracemalloc

tracemalloc.start()

# ... run your workload ...

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")
for stat in top_stats[:10]:
    print(stat)

Les sources courantes de croissance inattendue de la mémoire incluent :

• Appel fetchall() à une requête qui renvoie des millions de lignes. Utilisez fetchmany() ou un générateur à la place.
• Mise en cache des résultats de requêtes sans maxsize ni TTL. Les caches croissent jusqu’à ce que le processus soit relancé.
• Créer des curseurs en boucle sans les fermer. Chaque curseur ouvert contient son ensemble de résultats en mémoire.

Optimisation des index et des plans de requête

Vérifier la performance des requêtes côté serveur

Utilisez SET STATISTICS TIME ON et SET STATISTICS IO ON voyez combien de temps prennent les requêtes sur le serveur et combien de données elles lisent. Des lectures logiques élevées indiquent généralement un indice manquant. Exécutez ces instructions dans SQL Server Management Studio ou l’extension MSSQL pour Visual Studio Code, où la sortie apparaît dans le volet Messages :

SET STATISTICS TIME ON;
SET STATISTICS IO ON;

SELECT * FROM Production.Product WHERE ProductSubcategoryID = 1;

SET STATISTICS TIME OFF;
SET STATISTICS IO OFF;

Vous devriez voir un résultat similaire à :

Table 'Product'. Scan count 1, logical reads 3
SQL Server Execution Times: CPU time = 0 ms, elapsed time = 1 ms.

Si vous constatez des lectures logiques élevées ou des analyses de tables, envisagez d’ajouter un index.

Utilisez les indices de requête comme solution tactique

Les indications de requête remplacent les choix de l’optimiseur de requêtes en matière d’index et de stratégie de jointure. En production, ils sont précieux comme correctif rapide et à faible risque lorsqu’une requête régresse soudainement. Vous pouvez déployer immédiatement l’indice dans votre code applicatif pour stabiliser la requête pendant que vous enquêtez sur la cause profonde (index manquants, statistiques obsolètes ou changements de schéma).

Évitez de laisser des indices en place de façon permanente. Lorsque la distribution des données ou le schéma change, un indice codé en dur peut aggraver les choses. Considérez-les comme temporaires et revenez-y une fois le problème sous-jacent résolu :

cursor.execute("""
    SELECT * FROM Production.Product WITH (INDEX(AK_Product_Name))
    WHERE ProductSubcategoryID = %(subcategory_id)s
""", {"subcategory_id": 1})

Utilisez OPTION (RECOMPILE) pour contourner les plans mis en cache défectueux

SQL Server met en cache les plans de requête en fonction du premier ensemble de valeurs de paramètres qu’il voit. Si la distribution des données varie considérablement entre les appels, le plan mis en cache peut mal fonctionner pour certaines valeurs. Ce problème, appelé reniflage de paramètres, se manifeste souvent par une requête qui s’exécutait rapidement auparavant, mais qui met soudainement plusieurs secondes, voire plusieurs minutes, à s’exécuter.

OPTION (RECOMPILE)force SQL Server à construire un nouveau plan pour chaque exécution, ce qui constitue une solution immédiate efficace que vous pouvez déployer sans aucun changement côté serveur. Le compromis est un faible coût de compilation par appel, mais pour les requêtes qui s’exécutent peu fréquemment ou qui retournent des ensembles de résultats de taille variable, ce coût est négligeable comparé à l’exécution d’un mauvais plan.

Une fois le problème stabilisé, vous pouvez prendre votre temps pour appliquer une solution permanente comme réécrire la requête, ajouter des index filtrés ou utiliser des guides de plan :

cursor.execute("""
    SELECT * FROM Sales.SalesOrderHeader
    WHERE OrderDate > %(start_date)s
    OPTION (RECOMPILE)
""", {"start_date": start_date})

analyse des performances.

Chronométrez vos requêtes

Pour trouver des opérations lentes, enveloppez les requêtes avec time.perf_counter():

import time

start = time.perf_counter()
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": customer_id})
rows = cursor.fetchall()
elapsed = time.perf_counter() - start

print(f"Query returned {len(rows)} rows in {elapsed:.3f}s")

Pour une vision plus large de l'endroit où votre application passe du temps, utilisez le module intégré cProfile à Python :

python -m cProfile -s cumtime my_app.py

Cette vue montre le temps cumulé par appel de fonction, ce qui vous aide à identifier si la lenteur est liée à l’exécution des requêtes, au traitement des données ou à la latence réseau.

Utilisez Magasin des requêtes pour l’analyse côté serveur

Le timing côté client indique combien de temps prend une requête du point de vue de votre application, mais il combine la latence réseau, le temps d’exécution du serveur et le traitement client. Magasin des requêtes capture les plans d’exécution et les statistiques d’exécution sur le serveur, vous permettant de voir exactement comment SQL Server a exécuté chaque requête, à quelle fréquence elle s’exécutait et comment ses performances ont évolué au fil du temps.

Magasin des requêtes est particulièrement utile pour identifier l’analyse des paramètres, les régressions de plans et les requêtes qui consomment le plus de ressources serveur. Vous pouvez interroger directement les vues sys.query_store_runtime_stats et sys.query_store_plan, ou utiliser les rapports intégrés de Magasin des requêtes dans SQL Server Management Studio.

Utilisez les rapports du tableau de bord de performance

Les rapports du tableau de bord de performance dans SQL Server Management Studio fournissent un aperçu en temps réel de l’état de SQL Server, incluant les types d’attente actuels, les requêtes actives et coûteuses et les tendances CPU/IO. Utilisez-les pour repérer rapidement les goulets d’étranglement sans adresser directement de demandes aux DMV.

Liste de contrôle de performance

Connexion

• [ ] Activez le pooling de connexions.
• [ ] Ajustez la piscine pour votre charge de travail.
• [ ] Réutiliser les connexions dans les opérations.
• [ ] Gardez les connexions ouvertes dans les services de longue date.

Queries

• [ ] Sélectionnez uniquement les colonnes dont vous avez besoin.
• [ ] Utilisez la méthode de récupération appropriée pour chaque requête.
• [ ] Implémentez la pagination côté serveur.
• [ ] Définissez SET NOCOUNT ON une fois après la connexion.
• [ ] Minimisez les allers-retours en regroupant les requêtes.

Inserts

• [ ] Utilisez execute() pour les insertions sur une seule ligne.
• [ ] Utilisez executemany() pour des lots de petite à moyenne taille (~10-1 000 lignes).
• [ ] À utiliser bulkcopy() lorsque le débit compte plus que le contrôle par rangée.

Mise en cache

• [ ] Cache les données de référence avec un TTL pour éviter de servir des résultats obsolètes.

Resources

• [ ] Traiter les gros résultats en blocs ou avec des générateurs.
• [ ] Nettoyez les connexions rapidement.
• [ ] Surveille l’utilisation de la mémoire avec tracemalloc dans les services de longue durée d’exécution.

Contenu connexe

• Regroupement de connexions
• Opérations de copie en bloc
• Récupération des données