Optimalizace výkonu aplikací mssql-python

Ovladač mssql-python poskytuje několik funkcí a vzorů pro optimalizaci výkonu aplikací SQL Server, včetně poolování spojů, optimalizace dotazů a hromadných operací.

Správa připojení

Používejte sdružování připojení

Sdružování připojení je integrováno. Když voláte conn.close(), spojení se vrací do poolu k opětovnému použití místo toho, aby bylo zničeno, takže další connect() volání přeskočí drahé handshake:

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()

Konfigurujte velikost poolu pro workload

Upravte velikost poolu podle požadavků na souběžnost. Pokud vaše aplikace zpracovává více současných uživatelů, zvyšte tento pool. Pro lehčí pracovní zátěže menší fond šetří serverové zdroje:

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
)

Opětovné použití spojení v rámci operací

Otevírání nového spojení pro každý dotaz přidává režijní zátěž i při poolingu. Místo toho držte jedno spojení po dobu trvání logické operace:

# 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()

Udržujte spojení otevřená v dlouhodobě fungujících službách

Webové servery, frontové pracovníky a plánované úlohy, které běží nepřetržitě, by měly udržovat připojení otevřené, místo aby se připojovaly a odpojovaly při každé operaci. Otevření spojení zahrnuje TCP handshake, TLS vyjednávání a autentizaci, což může trvat 50–200 ms v závislosti na vzdálenosti sítě a způsobu autentizace. Pro pracovníka ve frontě zpracovávajícího tisíce zpráv za hodinu se tato režie rychle sčítá.

Udržujte spojení otevřené po celou dobu života pracovníka a znovu se připojte, když připojení selže. Spánek mezi iteracemi, abyste nemuseli server přetěžovat, když je fronta prázdná:

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()

Je-li povoleno sdružování připojení (ve výchozím nastavení), fond připojení spravuje nečinná připojení za vás. Ale pokud vypnete sdílení připojení nebo použijete jedno vyhrazené připojení, nastavte v připojovacím řetězci hodnoty Connection Timeout a Command Timeout, aby bylo možné zastaralá připojení odhalit včas, místo aby se aplikace zasekla.

Optimalizace dotazů

Fetch pouze potřebovala data

Výběr pouze sloupců, které vaše aplikace používá, snižuje přenos sítě, spotřebu paměti a dobu vykonávání dotazů.

# 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})

Použijte vhodné metody načítání

Ovladač poskytuje několik metod načítaní. Použijte ten, který odpovídá velikosti vašeho výsledku:

• fetchval() vrací jednu skalární hodnotu s minimálními režijními náklady.
• fetchall() Celou sadu výsledků se načte do paměti, což dobře funguje pro malé tabulky.
• fetchmany(n) Řádky se načítají v dávkách, přičemž u velkých sad výsledků zůstává využití paměti konstantní.

Vhodná velikost dávky pro fetchmany() závisí na šířce řádku. U úzkých řádků (několik malých sloupců, přibližně 1 KB každý) 1 000 řádků udržuje každou dávku přibližně 1 MB paměti. Pro širší řádky s velkými řetězci nebo sloupci s binárními daty použijte menší velikost dávky. Začněte s 1 000 a upravte podle svých dat.

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)

Použijte stránkování na straně serveru

Místo načítání všech řádků a jejich ořezávání v Pythonu použijte OFFSET/FETCH NEXT k načtení pouze stránky, kterou potřebujete.

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()

Použijte SET NOCOUNT ON

Ve výchozím nastavení SQL Server po každém DML příkazu odesílá zprávu "řádky postižené". SET NOCOUNT ON potlačuje tyto zprávy a snižuje síťový provoz. Je to nastavení na úrovni relace, takže ho nastavte jednou po připojení, místo abyste ho vkládali do každého dotazu.

# 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"}
)

Vyberte správnou metodu vložení

Ovladač poskytuje tři způsoby vložení dat, z nichž každý je přizpůsoben jiné škále:

Metoda	Počet řádků	Proč
execute()	1 řádek na každé volání	Použijte pro operace s jedním řádkem, jako je odesílání formulářů nebo API handlery, kde potřebujete vložené ID okamžitě.
executemany()	~10-1 000 řádků	Používá vazbu parametrů po sloupcích pro lepší propustnost než smyčka. Každý řádek odešle jako parametrizovaný příkaz.
bulkcopy()	Stovky řádků a více	Používá protokol TDS pro hromadné vkládání, který je výrazně efektivnější než vkládání po jednotlivých řádcích. Nejlepší pro načítání dat, migrace a dávkové zpracování.

Pro více podrobností a příkladů viz Načítání dat a pohybové vzory.

Jednotlivá vložení pomocí execute()

Použijte pro jednorázové vložení, když potřebujete okamžitý výsledek. Production.Product má několik sloupců NOT NULL bez výchozích nastavení, takže vložka je všechny uvádí:

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()

Dávkové vkládání pomocí executemany()

executemany() svazuje parametry po sloupcích a efektivně je odesílá. Používejte ho pro střední dávky místo volání execute() v cyklu. Všimněte si, že executemany() vyžaduje ? polohové značky se seznamem n-tic, zatímco execute() podporuje jak ? parametry, tak pojmenované %(name)s parametry pomocí diktů. Podrobnosti o každém stylu najdete v sekci Parametrizované dotazy .

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()

Hromadné kopírování pro velké objemy dat

Když je propustnost důležitější než řízení na řádku, přepněte na bulkcopy(). Přenáší řádky prostřednictvím protokolu TDS pro hromadné vkládání a vyhýbá se režii parametrizovaných příkazů pro jednotlivé řádky. Přesný bod, od kterého je bulkcopy() rychlejší než executemany(), závisí na šířce řádku a latenci sítě, ale obvykle se pohybuje v řádu nižších stovek řádků. Pro velmi malé dávky je executemany() jednodušší, protože bulkcopy() vytváří samostatné interní připojení a automaticky potvrzuje transakci.

Na rozdíl od execute() a executemany() mapuje bulkcopy() hodnoty na sloupce podle pozice, nikoli podle seznamu sloupců INSERT. Předejte column_mappings, chcete-li určit cílové sloupce, do nichž načítáte data, aby zdrojové n-tice odpovídaly správným sloupcům namísto prvního identitního sloupce tabulky:

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

Pro velmi velké objemy dat použijte generátor, abyste se vyhnuli načítání celé datové sady do paměti, a nastavte batch_size tak, aby změny průběžně potvrzoval:

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,
)

Strategie ukládání do mezipaměti

U referenčních dat, která se téměř nemění, (kategorie, tabulky vyhledávání, konfigurace), ukládejte výsledky do cache ve své aplikaci místo dotazování na každý požadavek.

functools.lru_cache v Pythonu poskytuje jednoduchou memoizaci, ale výsledky ukládá do mezipaměti neomezeně dlouho, dokud se proces nerestartuje. Pokud se základní data mohou měnit, použijte cachetools.TTLCache automatické obnovení po uplynutí časového limitu:

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()

Optimalizace sítě

Minimalizujte zpáteční jízdy

Každý dotaz je síťová cesta zpět k serveru. Spojte související dotazy do jedné dávky a použijte nextset() k procházení sad výsledků:

# 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()

Použití zpracování na straně serveru pro složitou logiku

Přesuňte agregaci a filtrování do SQL Serveru namísto načítání nezpracovaných řádků a jejich zpracování v Pythonu. Server vrací jeden souhrnný řádek místo potenciálně tisíců detailních řádků:

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})

Vyhněte se operacím s prokládaným kurzorem

Ovladač mssql-python nepodporuje více aktivních výsledkových sad (MARS). Pouze jeden kurzor může mít aktivní dotaz na každé připojení. Před spuštěním dalšího dotazu načtěte první množinu výsledků úplně, nebo použijte druhé spojení:

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()

Správa paměti

Zpracování velkých výsledků v částech

Načtení tabulky o více milionech řádků do seznamu spotřebovává paměť úměrně celé sadě výsledků. Použijte OFFSET a FETCH NEXT pro stránkování dat na straně serveru a zpracování jednoho bloku dat po druhém.

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
)

Používejte generátory pro streamování

Obalování fetchmany() generátoru v Python udržuje konstantní využití paměti bez ohledu na velikost tabulky. Volající iteruje řádek po řádku, aniž by načetl celou sadu výsledků. Pro extra velký zdroj kombinujte tabulky s UNION ALL a streamujte kombinovaný výsledek stejným způsobem.

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")

Rychle uklidit zdroje

Neuzavřená připojení zatěžují serverové zdroje a mohou vyčerpat kapacitu připojení. Použijte správce kontextu, abyste zajistili úklid i při výjimek.

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()

Monitorování využití paměti

Velké sady výsledků, dlouhotrvající cache a objekty připojení všechny spotřebovávají paměť. Pokud vaše aplikace běží jako služba, úniky paměti z neuzavřených kurzorů nebo neomezených caches mohou nakonec způsobit, že proces bude ukončen operačním systémem nebo kontejnerovým runtime.

Pomocí modulu tracemalloc jazyka Python pořiďte snímek paměti a najděte největší alokace.

import tracemalloc

tracemalloc.start()

# ... run your workload ...

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

Mezi běžné zdroje neočekávaného růstu paměti patří:

• Volání fetchall() u dotazu, který vrací miliony řádků. Použijte místo toho fetchmany() nebo generátor.
• Ukládání výsledků dotazů do mezipaměti bez maxsize nebo TTL. Cache rostou, dokud není proces restartován.
• Vytváření kurzorů v smyčce bez jejich zavírání. Každý otevřený kurzor uchovává množinu výsledků v paměti.

Optimalizace indexového a dotazového plánu

Zkontrolujte výkon serverových dotazů

Použijte SET STATISTICS TIME ON a SET STATISTICS IO ON, abyste viděli, jak dlouho trvají dotazy na serveru a kolik dat načtou. Vysoké logické hodnoty obvykle znamenají chybějící index. Spusť tyto příkazy v SQL Server Management Studio nebo v rozšíření MSSQL pro Visual Studio Code, kde se výstup objeví v panelu Zprávy:

SET STATISTICS TIME ON;
SET STATISTICS IO ON;

SELECT * FROM Production.Product WHERE ProductSubcategoryID = 1;

SET STATISTICS TIME OFF;
SET STATISTICS IO OFF;

Měl by se zobrazit výstup jako:

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

Pokud zaznamenáte vysoký počet logických čtení nebo prohledávání tabulek, zvažte přidání indexu.

Použijte dotazovací hinty jako taktické řešení

Nápovědy dotazu přebíjejí volby optimalizátoru dotazu týkající se indexů a strategií spojování. V produkčním prostředí jsou cenné jako rychlá oprava s nízkým rizikem, když se výkon dotazu náhle zhorší. Hint můžete nasadit přímo do aplikačního kódu, abyste stabilizovali dotaz, zatímco zkoumáte příčinu (chybějící indexy, zastaralé statistiky nebo změny schématu).

Vyhněte se trvalému zanechávání nápověd. Když se změní rozložení dat nebo schéma, pevně zakódovaná nápověda může situaci zhoršit. Považujte je za dočasné a znovu je zvažte poté, co bude základní problém vyřešen:

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

Použijte OPTION (RECOMPILE) k obejití špatných cacheovaných plánů

SQL Server ukládá plány dotazů na základě první sady hodnot parametrů, které vidí. Pokud se rozložení dat mezi hovory výrazně liší, může cacheovaný plán u některých hodnot podávat špatné výsledky. Tento problém, nazývaný parameter sniffing, se často projevuje jako dotaz, který "dříve byl rychlý", ale najednou trvá sekundy nebo minuty.

OPTION (RECOMPILE)nutí SQL Server vytvořit nový plán pro každé vykonání, což je efektivní okamžité řešení, které můžete nasadit bez jakýchkoli změn na straně serveru. Nevýhodou jsou sice malé náklady na kompilaci při každém volání, ale u dotazů, které se spouštějí zřídka nebo vracejí proměnlivě velké výsledkové sady, jsou tyto náklady zanedbatelné ve srovnání se spuštěním neefektivního plánu.

Jakmile se vám podaří problém stabilizovat, můžete si v klidu dát čas na zavedení trvalé opravy, například přepsání dotazu, přidání filtrovaných indexů nebo použití vodítek plánu:

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

Monitorování výkonu

Naplánujte své dotazy

Pro nalezení pomalých operací zabalte dotazy pomocí 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")

Pro širší přehled o tom, kde vaše aplikace tráví čas, použijte vestavěný cProfile modul Python:

python -m cProfile -s cumtime my_app.py

Tento pohled ukazuje kumulativní čas na volání funkce, což vám pomáhá určit, zda je zpomalení způsobeno prováděním dotazů, zpracováním dat nebo latencí sítě.

Použijte Query Store pro analýzu na straně serveru

Časování na straně klienta vám říká, jak dlouho dotaz trvá z pohledu vaší aplikace, ale kombinuje latenci sítě, dobu spuštění serveru a klientské zpracování. Query Store zaznamenává plány provádění a statistiky běhu na serveru, takže můžete přesně vidět, jak SQL Server vykonával každý dotaz, jak často běžel a jak se jeho výkon v průběhu času měnil.

Query Store je zvláště užitečný pro identifikaci sniffingu parametrů, regrese plánů a dotazů, které spotřebovávají nejvíce serverových zdrojů. Můžete se přímo dotazovat na zobrazení sys.query_store_runtime_stats a sys.query_store_plan nebo použít integrované sestavy Query Store v nástroji SQL Server Management Studio.

Používejte reporty výkonnostních dashboardů

Zprávy o výkonu v SQL Server Management Studio poskytují aktuální přehled o stavu SQL Server, včetně aktuálních typů čekání, aktivních drahých dotazů a trendů CPU/IO. Použijte je k rychlému odhalení úzkých míst, aniž byste přímo psali dotazy na DMV.

Kontrolní seznam výkonu

Připojení

• [ ] Povolte sdružování připojení.
• [ ] Nastavte velikost poolu pro vaši pracovní zátěž.
• [ ] Znovu použijte spojení v rámci operací.
• [ ] Udržujte spojení otevřená v dlouho běžících službách.

Queries

• [ ] Vyberte jen ty sloupce, které potřebujete.
• [ ] Použijte příslušnou metodu načtení pro každý dotaz.
• [ ] Implementujte stránkování na straně serveru.
• [ ] Nastavte SET NOCOUNT ON pouze jednou po připojení.
• [ ] Minimalizujte zpáteční cesty dávkováním dotazů.

Vložení

• [ ] Použijte execute() pro vkládání jednoho řádku.
• [ ] Použití executemany() pro malé až střední várky (~10–1 000 řádků).
• [ ] Použijte bulkcopy(), když je propustnost důležitější než kontrola jednotlivých řádků.

Cacheování

• [ ] Cache reference dat pomocí TTL, aby se zabránilo podávání zastaralých výsledků.

Resources

• [ ] Zpracovávejte velké výsledky po částech nebo pomocí generátorů.
• [ ] Rychle uklidněte spojení.
• [ ] Monitorujte využití paměti pomocí tracemalloc v dlouhodobě běžících službách.

Související obsah

• Sdružování připojení
• Operace hromadného kopírování
• Získávání dat