Obiekty wierszowe w mssql-python

Sterownik mssql-python zwraca dane wierszowe jako Row obiekty z operacji pobierania. Te obiekty zapewniają elastyczne wzorce dostępu:

  • Dostęp do atrybutów (row.ColumnName) jest najbardziej czytelnym wyborem dla kolumn nazwanych. Używaj go, gdy Twoje zapytanie ma znaną, stabilną listę kolumn.
  • Dostęp do klucza łańcuchowego (row['ColumnName']) zapewnia dostęp do kolumn według nazwy, co jest przydatne, gdy potrzebujesz programowego wyszukiwania kolumn lub gdy nazwy kolumn zawierają spacje lub znaki specjalne.
  • Dostęp do indeksu (row[0]) jest przydatny do zapytań dynamicznych, gdzie nazwy kolumn nie są znane podczas tworzenia lub podczas przetwarzania SELECT * wyników.
  • Rozpakowanie krotk (a, b, c = row) jest najbardziej zwięzłym wyborem dla ciał pętli z małą, stałą liczbą kolumn.
  • Dostęp mapujący (row._mapping) zwraca tylko do odczytu mapowanie nazw kolumn na wartości. Użyj go do konwersji wiersza do słownika lub przekazania wiersza kodowi, który oczekuje mapowania.

Dostęp do atrybutów

Uzyskaj bezpośredni dostęp do wartości kolumn według nazwy:

import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()

print(row.ProductID)     # 1
print(row.Name)          # 'Adjustable Race'
print(row.ListPrice)     # 0.00

Uwzględnij wielkość liter

Dostęp do nazw kolumn jest zależny od wielkości liter i odpowiada nazwam kolumn zwracanym przez SQL Server. Jeśli Twoja baza danych używa niespójnego okrewnictwa, użyj aliasów SQL AS do normalizacji nazw lub włącz ustawienia modułu lowercase (patrz Konfiguracja modułu):

cursor.execute("SELECT FirstName, LastName, EmailPromotion FROM Person.Person WHERE BusinessEntityID < 10")
row = cursor.fetchone()

print(row.FirstName)      # Works
print(row.LastName)       # Works
print(row.EmailPromotion) # Works
print(row.firstname)      # AttributeError - wrong case

Aliasy kolumn

Używaj aliasów SQL do tworzenia przyjaznych nazw atrybutów:

cursor.execute("""
    SELECT 
        p.ProductID,
        p.Name,
        c.Name AS Category
    FROM Production.Product p
    JOIN Production.ProductSubcategory c ON p.ProductSubcategoryID = c.ProductSubcategoryID
    WHERE p.ProductSubcategoryID IS NOT NULL
""")

for row in cursor:
    print(f"{row.Name} ({row.Category})")

Dostęp do indeksu

Wartości dostępu według indeksu kolumnowego opartego na zerze:

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()

print(row[0])  # ProductID
print(row[1])  # Name
print(row[2])  # ListPrice

Indeksowanie ujemne

Sterownik obsługuje negatywne indeksowanie w stylu Python:

cursor.execute("""
    SELECT Demo.A, Demo.B, Demo.C, Demo.D
    FROM (VALUES (1, 2, 3, 4)) AS Demo(A, B, C, D)
""")
row = cursor.fetchone()

print(row[-1])  # Last column (D)
print(row[-2])  # Second to last (C)

Dostęp do klucza łańcuchowego

Uzyskaj dostęp do wartości kolumn według nazwy za pomocą składni słownikowej:

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()

print(row['ProductID'])   # 1
print(row['Name'])        # 'Adjustable Race'
print(row['ListPrice'])   # Decimal('0.00')

Jest to przydatne, gdy nazwy kolumn zawierają spacje lub znaki specjalne, albo gdy trzeba programowo uzyskać dostęp do kolumn:

column_name = 'ListPrice'
value = row[column_name]  # Programmatic column access

Krojenie

Wyodrębniaj wiele wartości za pomocą wycinków:

cursor.execute("""
    SELECT Demo.A, Demo.B, Demo.C, Demo.D, Demo.E
    FROM (VALUES (1, 2, 3, 4, 5)) AS Demo(A, B, C, D, E)
""")
row = cursor.fetchone()

print(row[1:4])    # Columns B, C, D (indices 1, 2, 3)
print(row[:2])     # First two columns (A, B)
print(row[2:])     # From C to end

Rozpakowywanie krotek

Rozpakuj wartości wierszy bezpośrednio na zmienne:

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")

for product_id, name, price in cursor:
    print(f"#{product_id}: {name} - ${price}")

Częściowe rozpakowanie

Użycie * do przechwycenia pozostałych wartości:

cursor.execute("""
    SELECT TOP 2 ProductID, Name, ProductNumber, Color, Size, Weight
    FROM Production.Product
    WHERE ProductNumber IS NOT NULL
    ORDER BY ProductID
""")

for product_id, name, *rest in cursor:
    print(f"{product_id}: {name}, extra columns: {rest}")

Długość wiersza i iteracja

Pracuj z wymiarami wiersza i iteruj wartości kolumn:

Pobierz liczbę kolumn

Użyj len() do uzyskania liczby kolumn w rzędzie:

cursor.execute("SELECT * FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()

print(len(row))  # Number of columns

Iteruj nad wartościami

Przewijaj wartości kolumn w kolejności:

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")
row = cursor.fetchone()

for value in row:
    print(value)

Sprawdź, czy istnieje kolumna

Użyj hasattr() do testowania obecności nazwy kolumny:

# Use hasattr to check for column name
if hasattr(row, 'DiscountPrice'):
    print(f"Discount: {row.DiscountPrice}")
else:
    print("No discount available")

Przejście na typy wbudowane

Obiekty wierszowe można konwertować na standardowe typy Python do integracji z innymi bibliotekami i API:

Przekształć na krotki

Przekonwertuj wiersz na krotkę za pomocą tuple() konstruktora:

cursor.execute("SELECT ProductID, Name FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()

row_tuple = tuple(row)
print(row_tuple)  # (1, 'Adjustable Race')

Przekształtaj na listę

Przekonwertowanie wiersza na listę za pomocą list() konstruktora:

row_list = list(row)
print(row_list)  # [1, 'Adjustable Race']

Konwertuj do słownika

Przekonwertuj wiersz na słownik, gdy musisz go serializować do JSON, przekazać do silnika szablonowego lub połączyć z innymi danymi. Właściwość zwraca row._mapping tylko do odczytu mapowanie nazw kolumn na wartości w kolejności kolumn według zbioru wyników. Przekaż go do dict() zwykłego słownika:

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID = 1")
row = cursor.fetchone()

row_dict = dict(row._mapping)
print(row_dict)  # {'ProductID': 1, 'Name': 'Adjustable Race', 'ListPrice': Decimal('0.0000')}

Własność ma _mapping następujące zachowania:

  • Mapowanie jest tylko do odczytu. Przypisanie do klucza podnosi TypeError. Przelicz go najpierw na " dict() first", jeśli musisz zmienić wartości.
  • Klucze to nazwy kolumn z cursor.description. Wyszukiwania kluczowych są rozróżniane na wielka litera. Jeśli włączysz ustawienie lowercase modułu, klawisze są małymi literami, a mapowanie odpowiada tylko tym małym klawiszom. Dostęp do atrybutów i kluczy łańcuchowych w samym wierszu staje się niewrażliwy na wielka literę, gdy lowercase jest włączony, ale row._mapping wyszukiwania już nie.
  • Jeśli zbiór wyników zawiera zduplikowane nazwy kolumn, odwzorowanie zachowuje ostatnią wartość dla każdej nazwy. Użyj aliasów Transact-SQL (T-SQL), AS aby nadać każdej kolumnie unikalną nazwę lub odczytuj wartości według indeksu.
  • Własność ma pierwszeństwo przed kolumną wyników o nazwie _mapping, więc row._mapping zawsze zwraca odwzorowanie. Przeczytaj tę kolumnę z row['_mapping'].

Funkcja pomocnicza do konwersji dyktów

Stwórz wielokrotnego użytku funkcję pomocniczą, która konwertuje wszystkie pobrane wiersze na słowniki:

def rows_to_dicts(cursor):
    """Convert fetched rows to list of dictionaries."""
    return [dict(row._mapping) for row in cursor.fetchall()]

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")
products = rows_to_dicts(cursor)
for p in products:
    print(p['Name'])

Praca z wartościami zerowymi

Sterownik zwraca wartości NULL jako Python None:

cursor.execute("SELECT FirstName, MiddleName, LastName FROM Person.Person WHERE BusinessEntityID = 1")
row = cursor.fetchone()

if row.MiddleName is None:
    full_name = f"{row.FirstName} {row.LastName}"
else:
    full_name = f"{row.FirstName} {row.MiddleName} {row.LastName}"

Współpraca z kursor.description

Uzyskaj dostęp do metadanych kolumn obok danych wierszy:

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 5")

# Column information
for col in cursor.description:
    print(f"Name: {col[0]}, Type: {col[1]}")

# Fetch with metadata
row = cursor.fetchone()
for i, col in enumerate(cursor.description):
    print(f"{col[0]}: {row[i]}")

Często używane wzorce

Oto praktyczne wzorce pracy z obiektami wierszowymi w zastosowaniach rzeczywistych:

Wiersze procesów z dostępem nazwanym

Wiersze procesu korzystające z dostępu do atrybutów dla czytelnego, łatwego do utrzymania kodu:

def process_orders(conn):
    cursor = conn.cursor()
    cursor.execute("""
        SELECT SalesOrderID, CustomerID, OrderDate, TotalDue 
        FROM Sales.SalesOrderHeader 
        WHERE Status = 5
    """)
    
    for order in cursor:
        print(f"Order #{order.SalesOrderID}")
        print(f"  Customer: {order.CustomerID}")
        print(f"  Date: {order.OrderDate}")
        print(f"  Total: ${order.TotalDue:.2f}")

Buduj obiekty z wierszy

W zastosowaniach z modelem domenowym mapuj wiersze na klasy danych lub typowe obiekty. To mapowanie zapewnia automatyczne uzupełnianie IDE, sprawdzanie typów oraz wyraźną granicę między wierszami bazy danych a logiką aplikacji.

from dataclasses import dataclass
from datetime import date
from decimal import Decimal

@dataclass
class Product:
    id: int
    name: str
    price: Decimal
    created: date

def get_products(conn) -> list[Product]:
    cursor = conn.cursor()
    cursor.execute("SELECT ProductID, Name, ListPrice, SellStartDate FROM Production.Product WHERE ProductID < 10")
    
    return [
        Product(
            id=row.ProductID,
            name=row.Name,
            price=row.ListPrice,
            created=row.SellStartDate
        )
        for row in cursor
    ]

Eksport do JSON

Serializuj obiekty wierszowe do JSON z własnym obsługą typów dla wartości datetime i Decimal:

import json
from datetime import date, datetime
from decimal import Decimal

def json_serializer(obj):
    """Custom serializer for non-JSON types."""
    if isinstance(obj, (date, datetime)):
        return obj.isoformat()
    if isinstance(obj, Decimal):
        return float(obj)
    raise TypeError(f"Type {type(obj)} not serializable")

def export_to_json(cursor, filename):
    rows = [dict(row._mapping) for row in cursor.fetchall()]
    
    with open(filename, 'w') as f:
        json.dump(rows, f, default=json_serializer, indent=2)

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")
export_to_json(cursor, "products.json")

Agregacja w grupy

Grupuj wiersze według wartości kolumny i zbieraj je do słownika do analizy lub wyświetlania:

from collections import defaultdict

cursor.execute("""
    SELECT c.Name AS CategoryName, p.Name AS ProductName, p.ListPrice 
    FROM Production.Product p
    JOIN Production.ProductSubcategory c ON p.ProductSubcategoryID = c.ProductSubcategoryID
    WHERE p.ProductSubcategoryID IS NOT NULL
    ORDER BY c.Name
""")

products_by_category = defaultdict(list)
for row in cursor:
    products_by_category[row.CategoryName].append({
        'name': row.ProductName,
        'price': row.ListPrice
    })

for category, products in products_by_category.items():
    print(f"\n{category}:")
    for p in products:
        print(f"  - {p['name']}: ${p['price']}")

Treści powiązane

  • Ostatnia aktualizacja o