Práce s binárními daty

Ovladač mssql-python podporuje binary, varbinary, a image datové typy pro Microsoft SQL.

Microsoft SQL ukládá binární data do těchto typů sloupců:

Typ	Popis	Maximální velikost
binary(n)	Binární data s pevnou délkou.	8 000 bajtů
varbinary(n)	Binární data s proměnnou délkou	8 000 bajtů
varbinary(max)	Velká binární data.	2 GB
image	Původní velká binární data (zastaralá).	2 GB

Ovladač mssql-python vrací binární data jako Python bytes objekty.

Kdy uložit binární soubor do databáze versus do souborového systému:

• Ukládejte do databáze, když jsou soubory malé (pod 1 MB), transakční konzistence s ostatními daty je důležitá, nebo když potřebujete zálohovat data a soubory dohromady.
• Ukládejte do souborového systému (nebo Azure Blob Storage), když jsou soubory velké (přes 1 MB), potřebujete doručení přes CDN, nebo když potřebujete doručovat soubory přímo klientům bez návratu do databáze.
• Pro kompromis je funkce FILESTREAM od Microsoft SQL, která ukládá data do souborového systému s transakční konzistencí.

Vložte binární data

Předávejte objekty jazyka Python bytes jako parametry; ovladač je mapuje na sloupce varbinary.

Vložte bajty přímo

Vytvořte Python bytes objekt a předejte ho jako dotazovací parametr:

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

Vložit ze souboru

Přečtěte soubor v binárním režimu a vložte jeho obsah:

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

Vložit obrazový soubor

Ukládejte obrazové soubory s metadaty detekcí typů MIME z přípon souboru:

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

Načtení binárních dat

Ovladač vrací varbinární a binární sloupcové hodnoty jako objekty v Pythonbytes.

Sloupec pro načtení binárních souborů

Dotazujte binární sloupce a zkontrolujte vrácený 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

Uložení do souboru

Získejte binární data a zapíšte je na disk:

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

Export binárních řádků do souborů

Exportujte více binárních řádků do lokálních souborů:

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

Vložit binární hodnoty NULL

Předejte None, chcete-li vložit SQL NULL do binárního sloupce. Protože #Documents je dočasná tabulka, nejprve deklarujte typy parametrů pomocí setinputsizes(), aby ovladač navázal Content jako 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}
)

Note

Ovladač obvykle odvodí typy parametrů pomocí SQLDescribeParam, ale nemůže vyřešit metadata typů pro dočasnou tabulku nebo proměnnou tabulky. Vložení None do sloupce binary nebo varbinary dočasného objektu bez setinputsizes() vyvolá ProgrammingError: Implicit conversion from data type varchar to varbinary(max) is not allowed. Předejte jeden záznam na parametr v pořadí a použijte SQL typovou konstantu, například mssql_python.SQL_VARBINARY pro binární sloupec. U běžné (trvalé) tabulky ovladač automaticky určí typ a můžete předat None přímo.

Velká binární data

U souborů nad pár megabajtů vložte celý obsah do jednoho varbinárního (max) zápisu místo čtení souboru po malých částech.

Streamovat velké soubory

U velkých souborů zpracováváme v částech:

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

Použijte hromadnou kopii pro binární data

Použijte tuto metodu bulkcopy() k efektivnímu vložení více binárních souborů v jedné operaci.

Important

bulkcopy() data se načítají přes samostatné připojení k serveru, takže cílová tabulka musí již existovat a být commitovaná a viditelná pro ostatní relace. Pokud tabulku vytvoříte ve stejném skriptu s vypnutým automatickým commitem, zavolejte conn.commit() před bulkcopy(). Lokální dočasné tabulky (#name) nejsou podporovány, protože jsou soukromé pro relaci, která je vytvořila. Nepotvrzený nebo nedosažitelný cíl způsobí selhání bulkcopy() s vypršením časového limitu Failed to retrieve destination metadata.

Ve výchozím nastavení mapuje bulkcopy() každou hodnotu v řádku ke sloupci tabulky podle pořadí, takže každý sloupec musí mít hodnotu ve správném pořadí. Když má tabulka sloupec identity nebo vyplníte jen některé sloupce, přejděte column_mappings a explicitně pojmenujte cílové sloupce. Jinak se hodnoty přesunou do špatných sloupců a bulkcopy() selžou.

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ární datové operace

Použijte funkci Microsoft SQL HASHBYTES k porovnání binárního obsahu bez načítání plných hodnot.

Porovnejte binární data

Najděte binární řádky, které sdílejí identický obsah, výpočtem a porovnáním hashů obsahu v Microsoft SQL Server:

# 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

Výpočet hashe v Pythonu

Vypočítejte SHA-256 hashe v Python a uložte je spolu s binárními daty:

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

Kódování a dekódování binárních dat

Převádějte binární data na a z řetězců kódovaných base64:

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

Práce s konkrétními binárními formáty

Tyto příklady ukazují, jak ověřit a vložit běžné formáty souborů před jejich uložením.

Dokumenty PDF

Před vložením ověřte podpisy PDF souborů:

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

Obrázky s PIL/Pillow

Upravte velikost obrázků pomocí Pillow (PIL) před uložením, abyste ušetřili místo:

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

Komprimovaná data

Snižte úložiště kompresí binárních dat pomocí gzip před vložením:

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)

Osvědčené postupy

Použijte tyto pokyny k výběru správného typu sloupce a strategie ukládání binárních dat.

Používejte vhodné typy sloupců

Vyberte typ sloupce podle charakteristik vašich dat:

Pro různé binární případy užití vyberte příslušný typ dat Microsoft SQL:

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

Zvažte FILESTREAM pro velké soubory

U velkých souborů (přes 1 MB) zvažte funkci Microsoft SQL FILESTREAM, která ukládá data do souborového systému. FILESTREAM vyžaduje konfiguraci na straně serveru před použitím:

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

Validace binárních dat

Ověřte typy souborů kontrolou magických bajtů (podpisů souboru) před uložením binárních dat:

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

Související obsah

• Mapování datových typů
• hromadné kopírování
• Zpracovat hodnoty NULL