Arbeit mit Binärdaten

Der mssql-python-Treiber unterstützt binary, varbinary, und image Datentypen für Microsoft SQL.

Microsoft SQL speichert Binärdaten in diesen Spaltentypen:

Der mssql-python-Treiber liefert binäre Daten als Python-Objekte bytes zurück.

Wann Binärdateien in der Datenbank im Vergleich zum Dateisystem gespeichert werden:

• In der Datenbank speichern, wenn Dateien klein sind (unter 1 MB), transaktionale Konsistenz mit anderen Daten wichtig ist oder man Daten und Dateien zusammen sichern muss.
• Speichere im Dateisystem (oder Azure Blob Storage), wenn Dateien groß sind (über 1 MB), du eine CDN-Lieferung brauchst oder Dateien direkt an Clients ausliefern musst, ohne Datenbank-Roundtrips.
• Als Mittelweg speichert die FILESTREAM-Funktion von Microsoft SQL Daten im Dateisystem mit transaktionaler Konsistenz.

Binärdaten einfügen

Python-Objekte bytes als Parameter übergeben; der Treiber bildet sie auf varbinäre Spalten ab.

Bytes direkt einfügen

Erstelle ein Python-Objekt bytes und gib es als Abfrageparameter ein:

import mssql_python

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

# Create a table to hold binary documents
cursor.execute("""
    CREATE TABLE #Documents (
        ID INT IDENTITY PRIMARY KEY,
        Name NVARCHAR(255),
        Content VARBINARY(MAX),
        Size INT NULL,
        ContentHash VARBINARY(32) NULL
    )
""")

# Binary data as bytes
binary_data = b'\x00\x01\x02\x03\x04\x05'

cursor.execute(
    "INSERT INTO #Documents (Name, Content) VALUES (%(name)s, %(content)s)",
    {"name": "sample.bin", "content": binary_data}
)
conn.commit()

Aus Datei einfügen

Lesen Sie eine Datei im Binärmodus und fügen Sie ihren Inhalt ein:

def insert_file(cursor, file_path: str, name: str):
    """Insert a file as binary data."""
    with open(file_path, "rb") as f:
        content = f.read()
    
    cursor.execute(
        "INSERT INTO #Documents (Name, Content, Size) VALUES (%(name)s, %(content)s, %(size)s)",
        {"name": name, "content": content, "size": len(content)}
    )

insert_file(cursor, "image.png", "profile_picture.png")
conn.commit()

Bilddatei einfügen

Speichern Bilddateien mit Metadaten, indem MIME-Typen aus Dateiendungen erkannt werden:

# Create a table to hold images
cursor.execute("""
    CREATE TABLE #Images (
        ID INT IDENTITY PRIMARY KEY,
        FileName NVARCHAR(255),
        FileSize INT NULL,
        ContentType NVARCHAR(100) NULL,
        Width INT NULL,
        Height INT NULL,
        ImageData VARBINARY(MAX),
        Description NVARCHAR(MAX) NULL
    )
""")

def insert_image(cursor, image_path: str, description: str):
    """Insert an image into the database."""
    import os
    
    with open(image_path, "rb") as f:
        image_data = f.read()
    
    cursor.execute("""
        INSERT INTO #Images (FileName, FileSize, ContentType, ImageData, Description)
        VALUES (%(filename)s, %(filesize)s, %(content_type)s, %(data)s, %(desc)s)
    """, {
        "filename": os.path.basename(image_path),
        "filesize": len(image_data),
        "content_type": get_content_type(image_path),
        "data": image_data,
        "desc": description
    })

def get_content_type(path: str) -> str:
    """Determine MIME type from file extension."""
    ext = path.lower().split(".")[-1]
    types = {
        "png": "image/png",
        "jpg": "image/jpeg",
        "jpeg": "image/jpeg",
        "gif": "image/gif",
        "pdf": "application/pdf",
    }
    return types.get(ext, "application/octet-stream")

Abrufen von Binärdaten

Der Treiber liefert varibinäre und binäre Spaltenwerte als Python-Objekte bytes zurück.

Binärspalte abrufen

Fragen Sie die Binärspalten ab und überprüfen Sie das zurückgegebene bytes-Objekt:

cursor.execute(
    "SELECT LargePhotoFileName, LargePhoto FROM Production.ProductPhoto WHERE ProductPhotoID = %(id)s",
    {"id": 70}
)
row = cursor.fetchone()

# LargePhoto is bytes
print(type(row.LargePhoto))  # <class 'bytes'>
print(len(row.LargePhoto))   # Number of bytes

In Datei speichern

Binärdaten abrufen und auf Festplatte schreiben:

def save_photo(cursor, photo_id: int, output_path: str):
    """Retrieve binary data and save to file."""
    cursor.execute(
        "SELECT LargePhoto FROM Production.ProductPhoto WHERE ProductPhotoID = %(id)s",
        {"id": photo_id}
    )
    row = cursor.fetchone()
    
    if row and row.LargePhoto:
        with open(output_path, "wb") as f:
            f.write(row.LargePhoto)
        print(f"Saved {len(row.LargePhoto)} bytes to {output_path}")
    else:
        print("Photo not found or empty")

save_photo(cursor, 70, "output.gif")

Binärzeilen in Dateien exportieren

Exportiere mehrere binäre Zeilen in lokale Dateien:

import os

def export_photos(cursor, output_dir: str):
    """Export product photos to a directory."""
    os.makedirs(output_dir, exist_ok=True)
    
    cursor.execute("""
        SELECT ProductPhotoID, LargePhotoFileName, LargePhoto
        FROM Production.ProductPhoto
        WHERE ProductPhotoID > 1
    """)
    
    count = 0
    for row in cursor:
        output_path = os.path.join(output_dir, f"{row.ProductPhotoID}_{row.LargePhotoFileName}")
        with open(output_path, "wb") as f:
            f.write(row.LargePhoto)
        count += 1
    
    print(f"Exported {count} photos to {output_dir}")

export_photos(cursor, "./photos")

Fügen Sie NULL-Binärwerte ein

Übergeben Sie None, um einen SQL-NULL-Wert in eine Binärspalte einzufügen. Da #Documents eine temporäre Tabelle ist, deklarieren Sie die Parametertypen mit setinputsizes() erst, sodass der Treiber bindet Content als varbinary:

cursor.setinputsizes([(mssql_python.SQL_WVARCHAR, 255, 0), (mssql_python.SQL_VARBINARY, 0, 0)])
cursor.execute(
    "INSERT INTO #Documents (Name, Content) VALUES (%(name)s, %(content)s)",
    {"name": "empty", "content": None}
)

Große binäre Daten

Bei Dateien mit mehr als einigen Megabyte den gesamten Inhalt in einem einzigen varbinary(max)-Schreibvorgang einfügen, anstatt die Datei in kleinen Blöcken zu lesen.

Streame große Dateien

Verarbeiten Sie große Dateien in Blöcken:

def insert_large_file(conn, table: str, file_path: str, chunk_size: int = 8192):
    """Insert large file in chunks using updatetext-style approach."""
    # Validate table name to prevent SQL injection
    import re
    if not re.match(r'^#{0,2}[A-Za-z_][A-Za-z0-9_.]*$', table):
        raise ValueError(f"Invalid table name: {table}")

    cursor = conn.cursor()
    
    # Get file size
    import os
    file_size = os.path.getsize(file_path)
    
    # Insert initial row with empty binary
    cursor.execute(f"""
        INSERT INTO {table} (Name, Content, Size)
        VALUES (%(name)s, 0x, %(size)s);
        SELECT SCOPE_IDENTITY();
    """, {"name": os.path.basename(file_path), "size": file_size})
    
    row_id = cursor.fetchval()
    
    # For modern Microsoft SQL, better to use varbinary(max) and single insert
    # This example shows streaming approach
    with open(file_path, "rb") as f:
        content = f.read()
    
    cursor.execute(f"""
        UPDATE {table} SET Content = %(content)s WHERE ID = %(id)s
    """, {"content": content, "id": row_id})
    
    conn.commit()
    return row_id

Verwenden Sie Massenkopien für Binärdaten

Verwenden Sie die Methode bulkcopy() , um effizient mehrere Binärdateien in einer einzigen Operation einzufügen.

Standardmäßig ordnet bulkcopy() jeden Wert in einer Zeile anhand der Ordnungsposition einer Tabellenspalte zu, sodass für jede Spalte der Reihe nach ein Wert vorhanden sein muss. Wenn die Tabelle eine Identitätsspalte hat oder du nur einige Spalten ausfüllst, gehe über, column_mappings um die Zielspalten explizit zu benennen. Andernfalls verschieben sich die Werte auf die falschen Spalten und bulkcopy() scheitern.

def bulk_insert_files(conn, table: str, file_paths: list[str]):
    """Bulk insert multiple binary files."""
    import os

    data = []
    for path in file_paths:
        with open(path, "rb") as f:
            content = f.read()
        data.append((os.path.basename(path), content, len(content)))

    cursor = conn.cursor()
    result = cursor.bulkcopy(table, data, column_mappings=["Name", "Content", "Size"])
    conn.commit()
    return result["rows_copied"]

Binärdatenoperationen

Verwenden Sie die HASHBYTES Funktion von Microsoft SQL, um binäre Inhalte zu vergleichen, ohne vollständige Werte abzurufen.

Binärdaten vergleichen

Finden Sie binäre Zeilen, die identische Inhalte teilen, indem Sie Inhaltshashes in Microsoft SQL Server berechnen und vergleichen:

# Find rows with identical binary content by hash
cursor.execute("""
    SELECT LargePhotoFileName, HASHBYTES('SHA2_256', LargePhoto) AS PhotoHash
    FROM Production.ProductPhoto
    WHERE ProductPhotoID > 1
""")

hashes = {}
for row in cursor:
    hash_value = row.PhotoHash  # bytes
    if hash_value in hashes:
        print(f"Duplicate: {row.LargePhotoFileName} matches {hashes[hash_value]}")
    else:
        hashes[hash_value] = row.LargePhotoFileName

Hash in Python berechnen

Berechnen Sie SHA-256-Hashes in Python und speichern Sie sie zusammen mit Binärdaten:

import hashlib

def insert_with_hash(cursor, name: str, content: bytes):
    """Insert binary data with computed hash."""
    content_hash = hashlib.sha256(content).digest()
    
    cursor.execute("""
        INSERT INTO #Documents (Name, Content, ContentHash)
        VALUES (%(name)s, %(content)s, %(hash)s)
    """, {"name": name, "content": content, "hash": content_hash})

Binärdaten kodieren und dekodieren

Binärdaten in und von base64-codierten Strings konvertieren:

import base64

# Store base64-encoded string
def insert_base64(cursor, name: str, base64_data: str):
    """Insert base64-encoded data as binary."""
    binary_data = base64.b64decode(base64_data)
    cursor.execute(
        "INSERT INTO #Documents (Name, Content) VALUES (%(name)s, %(content)s)",
        {"name": name, "content": binary_data}
    )

# Retrieve as base64
def get_as_base64(cursor, doc_id: int) -> str:
    """Retrieve binary data as base64 string."""
    cursor.execute("SELECT Content FROM #Documents WHERE ID = %(id)s", {"id": doc_id})
    row = cursor.fetchone()
    return base64.b64encode(row.Content).decode("utf-8")

Arbeit mit spezifischen Binärformaten

Diese Beispiele zeigen, wie man gängige Dateiformate validiert und einfügt, bevor sie gespeichert werden.

PDF-Dokumente

Überprüfen Sie PDF-Dateisignaturen, bevor Sie sie einfügen:

def insert_pdf(cursor, pdf_path: str, title: str):
    """Insert a PDF document."""
    with open(pdf_path, "rb") as f:
        pdf_data = f.read()
    
    # Verify it's a PDF (magic bytes)
    if not pdf_data.startswith(b'%PDF'):
        raise ValueError("Not a valid PDF file")
    
    cursor.execute(
        "INSERT INTO #Documents (Name, Content) VALUES (%(title)s, %(data)s)",
        {"title": title, "data": pdf_data}
    )

Bilder mit PIL/Pillow

Vergrößern Sie Bilder mit Pillow (PIL), bevor Sie sie speichern, um Speicherplatz zu sparen:

from PIL import Image
import io
import os

def insert_resized_image(cursor, image_path: str, max_size: tuple = (800, 600)):
    """Insert a resized image."""
    # Open and resize
    img = Image.open(image_path)
    img.thumbnail(max_size, Image.LANCZOS)
    
    # Convert to bytes
    buffer = io.BytesIO()
    img.save(buffer, format=img.format or "PNG")
    image_bytes = buffer.getvalue()
    
    cursor.execute("""
        INSERT INTO #Images (FileName, Width, Height, ImageData)
        VALUES (%(name)s, %(width)s, %(height)s, %(data)s)
    """, {
        "name": os.path.basename(image_path),
        "width": img.width,
        "height": img.height,
        "data": image_bytes
    })

def get_image_as_pil(cursor, image_id: int) -> Image.Image:
    """Retrieve image as PIL Image object."""
    cursor.execute("SELECT ImageData FROM #Images WHERE ID = %(id)s", {"id": image_id})
    row = cursor.fetchone()
    return Image.open(io.BytesIO(row.ImageData))

Komprimierte Daten

Reduzieren Sie den Speicher durch Komprimieren von Binärdaten mit gzip vor der Einfügung:

import gzip

def insert_compressed(cursor, name: str, data: bytes):
    """Insert data with gzip compression."""
    compressed = gzip.compress(data)
    
    cursor.execute(
        "INSERT INTO #Documents (Name, Content, Size) VALUES (%(name)s, %(content)s, %(size)s)",
        {"name": name, "content": compressed, "size": len(data)}
    )

def get_decompressed(cursor, data_id: int) -> bytes:
    """Retrieve and decompress data."""
    cursor.execute(
        "SELECT Content FROM #Documents WHERE ID = %(id)s",
        {"id": data_id}
    )
    row = cursor.fetchone()
    return gzip.decompress(row.Content)

Bewährte Methoden

Wenden Sie diese Richtlinien an, um den richtigen Spaltentyp und die richtige Speicherstrategie für Binärdaten auszuwählen.

Verwenden Sie geeignete Spaltentypen

Wählen Sie den Spaltentyp basierend auf den Eigenschaften Ihrer Daten:

Für verschiedene binäre Anwendungsfälle wählen Sie den passenden Microsoft SQL-Datentyp aus:

-- For small fixed-size binary (for example, hashes, UUIDs)
binary(32)      -- SHA-256 hash

-- For variable-size binary up to 8KB (for example, thumbnails, small icons)
varbinary(8000)

-- For large binary data (for example, documents, images)
varbinary(max)  -- Up to 2GB

Betrachten wir FILESTREAM für große Dateien

Für große Dateien (über 1 MB) sollten Sie die Microsoft SQL FILESTREAM-Funktion in Betracht ziehen, die Daten im Dateisystem speichert. FILESTREAM benötigt vor der Nutzung eine serverseitige Konfiguration :

# FILESTREAM-enabled databases store large binaries more efficiently
# Access is still through normal queries but storage is file-based
large_binary_data = b"\x25\x50\x44\x46" + b"\x00" * 100  # sample data

cursor.execute("""
    CREATE TABLE ##FileStreamDemo (Name NVARCHAR(100), Document VARBINARY(MAX))
""")
cursor.execute("""
    INSERT INTO ##FileStreamDemo (Name, Document)
    VALUES (%(name)s, %(content)s)
""", {"name": "large_doc.pdf", "content": large_binary_data})

cursor.execute("SELECT Name, DATALENGTH(Document) AS DocSize FROM ##FileStreamDemo")
row = cursor.fetchone()
print(f"{row.Name}: {row.DocSize} bytes")

Binärdaten validieren

Validieren Sie Dateitypen, indem Sie magische Bytes (Dateisignaturen) vor dem Speichern von Binärdaten überprüfen:

def insert_safe_image(cursor, name: str, data: bytes):
    """Insert image with validation."""
    # Check file signatures (magic bytes)
    signatures = {
        b'\x89PNG': 'image/png',
        b'\xff\xd8\xff': 'image/jpeg',
        b'GIF87a': 'image/gif',
        b'GIF89a': 'image/gif',
    }
    
    content_type = None
    for sig, mime in signatures.items():
        if data.startswith(sig):
            content_type = mime
            break
    
    if content_type is None:
        raise ValueError("Unknown or unsupported image format")
    
    cursor.execute("""
        INSERT INTO #Images (FileName, ContentType, ImageData)
        VALUES (%(name)s, %(type)s, %(data)s)
    """, {"name": name, "type": content_type, "data": data})

Verwandte Inhalte

• Datentypzuordnungen
• Mehrfachkopie
• Mit NULL-Werten umgehen