Bekerja dengan data biner

Driver mssql-python mendukung binary, varbinary, dan image tipe data untuk Microsoft SQL.

Microsoft SQL menyimpan data biner dalam jenis kolom ini:

Jenis	Deskripsi	Ukuran maks
binary(n)	Data biner panjang tetap.	8.000 byte
varbinary(n)	Data biner panjang variabel.	8.000 byte
varbinary(max)	Data biner besar.	2 GB
image	Data biner besar lama (tidak digunakan lagi).	2 GB

Driver mssql-python mengembalikan data biner sebagai objek Pythonbytes.

Kapan harus menyimpan biner dalam database versus sistem file:

• Simpan dalam database saat file kecil (di bawah 1 MB), konsistensi transaksional dengan data lain penting, atau Anda perlu mencadangkan data dan file bersama-sama.
• Simpan di sistem file (atau Azure Blob Storage) saat file berukuran besar (lebih dari 1 MB), Anda memerlukan pengiriman CDN, atau Anda perlu menyajikan file langsung ke klien tanpa bolak-balik database.
• Untuk jalan tengah, fitur FILESTREAM Microsoft SQL menyimpan data dalam sistem file dengan konsistensi transaksional.

Menyisipkan data biner

Lewatkan objek Python bytes sebagai parameter; pengandar memetakannya ke kolom varbinary.

Sisipkan byte secara langsung

Buat objek Python bytes dan teruskan sebagai parameter kueri:

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

Sisipkan dari file

Baca file dalam mode biner dan masukkan isinya:

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

Sisipkan file gambar

Simpan file gambar dengan metadata dengan mendeteksi jenis MIME dari ekstensi file:

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

Mengambil data biner

Driver mengembalikan nilai kolom varbiner dan biner sebagai objek Pythonbytes.

Ambil kolom biner

Kueri kolom biner dan periksa objek yang ditampilkan bytes :

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

Simpan ke file

Ambil data biner dan tulis ke 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")

Ekspor baris biner ke file

Ekspor beberapa baris biner ke file lokal:

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

Sisipkan nilai biner NULL

Berikan None untuk menyisipkan SQL NULL ke kolom biner. Karena #Documents adalah tabel sementara, deklarasikan jenis parameter dengan menggunakan setinputsizes() terlebih dahulu sehingga driver mengikat Content sebagai 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

Driver biasanya menyimpulkan jenis parameter melalui SQLDescribeParam, tetapi tidak dapat menyelesaikan metadata jenis untuk tabel sementara atau variabel tabel. Menyisipkan None ke dalam kolom binary atau varbinary dari objek sementara tanpa setinputsizes() menimbulkan ProgrammingError: Implicit conversion from data type varchar to varbinary(max) is not allowed. Teruskan satu entri per parameter, secara berurutan, dan gunakan konstanta jenis SQL seperti mssql_python.SQL_VARBINARY untuk kolom biner. Untuk tabel reguler (permanen), driver menentukan tipe secara otomatis dan Anda dapat langsung meneruskan None.

Data biner besar

Untuk file lebih dari beberapa megabyte, masukkan konten lengkap dalam satu penulisan varbinary(max) alih-alih membaca file dalam potongan-potongan kecil.

Streamingkan file berukuran besar

Untuk file besar, proses dalam potongan:

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

Menggunakan salinan massal untuk data biner

Gunakan metode bulkcopy() untuk menyisipkan beberapa file biner secara efisien dalam satu operasi.

Penting

bulkcopy() memuat data melalui koneksi terpisah ke server, sehingga tabel tujuan harus sudah ada dan diterapkan serta terlihat oleh sesi lain. Jika Anda membuat tabel dalam skrip yang sama saat autocommit dimatikan, panggil conn.commit() sebelum bulkcopy(). Tabel sementara lokal (#name) tidak didukung karena bersifat pribadi untuk sesi yang membuatnya. Tujuan yang belum dikomitkan atau tidak dapat dijangkau menyebabkan bulkcopy() gagal dengan galat batas waktu Failed to retrieve destination metadata.

Secara default, bulkcopy() memetakan setiap nilai dalam baris ke kolom tabel berdasarkan posisi ordinal, sehingga setiap kolom harus memiliki nilai secara berurutan. Saat tabel memiliki kolom identitas atau Anda hanya mengisi beberapa kolom, teruskan column_mappings untuk memberi nama kolom target secara eksplisit. Jika tidak, nilainya bergeser ke kolom yang salah dan bulkcopy() gagal.

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

Operasi data biner

Gunakan fungsi Microsoft SQL HASHBYTES untuk membandingkan konten biner tanpa mengambil nilai penuh.

Bandingkan data biner

Temukan baris biner yang berbagi konten identik dengan menghitung dan membandingkan hash konten di 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

Menghitung hash di Python

Hitung hash SHA-256 di Python dan simpan bersama data biner:

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

Mengkodekan dan memecahkan kode data biner

Mengonversi data biner ke dan dari string yang dikodekan 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")

Bekerja dengan format biner tertentu

Contoh ini menunjukkan cara memvalidasi dan menyisipkan format file umum sebelum menyimpannya.

Dokumen PDF

Verifikasi tanda tangan file PDF sebelum memasukkannya:

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

Gambar dengan PIL/Pillow

Ubah ukuran gambar menggunakan Bantal (PIL) sebelum menyimpannya untuk menghemat ruang:

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

Data terkompresi

Kurangi penyimpanan dengan mengompresi data biner dengan gzip sebelum dimasukkan:

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)

Praktik terbaik

Terapkan panduan ini untuk memilih jenis kolom dan strategi penyimpanan yang tepat untuk data biner.

Menggunakan jenis kolom yang sesuai

Pilih jenis kolom berdasarkan karakteristik data Anda:

Untuk kasus penggunaan biner yang berbeda, pilih tipe data Microsoft SQL yang sesuai:

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

Pertimbangkan FILESTREAM untuk file besar

Untuk file besar (lebih dari 1 MB), pertimbangkan fitur Microsoft SQL FILESTREAM, yang menyimpan data dalam sistem file. FILESTREAM memerlukan konfigurasi sisi server sebelum digunakan:

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

Validasi data biner

Validasi jenis file dengan memeriksa byte ajaib (tanda tangan file) sebelum menyimpan data biner:

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

Konten terkait

• Pemetaan jenis data
• Salinan massal
• Menangani nilai NULL