İkili verilerle çalışma

mssql-python sürücüsü, Microsoft SQL için , binary, ve varbinary veri türlerini desteklerimage.

Microsoft SQL ikili verileri şu sütun tiplerinde saklar:

Türü	Açıklama	Maksimum boyut
binary(n)	Sabit uzunluklu ikili veri.	8.000 bayt
varbinary(n)	Değişken uzunlukta ikili veriler.	8.000 bayt
varbinary(max)	Büyük ikili veri.	2GB
image	Eski büyük ikili veri (kullanımdan kaldırıldı).	2GB

mssql-python sürücüsü, ikili verileri Python bytes nesneleri olarak döndürür.

İkili veriler ne zaman veritabanında, ne zaman dosya sisteminde saklanmalıdır:

• Dosyalar küçük olduğunda (1 MB'dan küçükse), diğer verilerle işlem tutarlılığı önemli olduğunda veya veri ile dosyaları birlikte yedeklemeniz gerektiğinde veritabanında saklayın.
• Dosyalar büyük olduğunda (1 MB'ın üzerinde), CDN üzerinden sunum gerektiğinde veya dosyaları veritabanına gidip gelmeden doğrudan istemcilere sunmanız gerektiğinde, bunları dosya sisteminde (veya Azure Blob Depolama'ta) depolayın.
• Orta yol olarak, Microsoft SQL'nin FILESTREAM özelliği, verileri dosya sisteminde işlemsel tutarlılıkla depolar.

İkili veri ekle

Python bytes nesnelerini parametre olarak ilet; sürücü bunları varbiner sütunlara eşler.

Baytları doğrudan ekle

Bir Python bytes nesnesi oluşturun ve bunu sorgu parametresi olarak geçirin:

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

Dosyadan ekle

Bir dosyayı ikili modda okuyun ve içeriğini ekleyin:

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

Resim dosyası ekle

Dosya uzantılarından MIME türlerini tespit ederek meta veri içeren görüntü dosyalarını depolayın:

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

İkili verileri alma

Sürücü, varbinary ve binary sütun değerlerini Python bytes nesneleri olarak döndürür.

İkili veri sütununu getir

İkili sütunları sorgulayın ve döndürülen bytes nesneyi inceleyin:

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

Dosyaya kaydet

İkili veri alın ve diske yazın:

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

İkili satırları dosyalara aktarın

Birden fazla ikili satırı yerel dosyalara aktarın:

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

NULL ikili değerleri ekleyin

Pass None ile bir SQL NULL'u ikili sütuna ekle. #Documents geçici bir tablo olduğundan, sürücünün Content öğesini varbinary olarak bağlaması için önce setinputsizes() kullanarak parametre türlerini bildirin:

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

Sürücü genellikle parametre tiplerini , üzerinden SQLDescribeParamçıkarır, ancak geçici bir tablo veya tablo değişkeni için tür meta verilerini çözemez. setinputsizes() içermeyen bir geçici nesnenin binary veya varbinary sütununa None eklemek, ProgrammingError: Implicit conversion from data type varchar to varbinary(max) is not allowed hatasına neden olur. Her parametre için sırayla bir değer iletin ve ikili sütun için mssql_python.SQL_VARBINARY gibi bir SQL veri türü sabiti kullanın. Normal (kalıcı) bir tabloda sürücü türü otomatik olarak belirler ve None öğesini doğrudan iletebilirsiniz.

Büyük ikili veri

Birkaç megabayttan uzun dosyalar için, dosyayı küçük parçalar halinde okumak yerine tam içeriği tek bir varbinary(max) yazıya ekleyin.

Büyük dosyaları akış

Büyük dosyalar için parçalar halinde işleyin:

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

İkili veri için toplu kopya kullanın

Bu yöntemi, tek bir işlemde birden fazla ikili dosyayı verimli bir şekilde eklemek için kullanın bulkcopy() .

Important

bulkcopy() sunucuya ayrı bir bağlantı üzerinden veri yükler, bu yüzden hedef tablo zaten var olmalı ve diğer oturumlara bağlı ve görünür olmalıdır. Aynı scriptte otomatik commit kapalıyken tablo oluşturursanız, conn.commit() önce bulkcopy()çağırın. Yerel geçici tablolar (#name) desteklenmez çünkü bunlar, onları oluşturan oturuma özeldir. Onaylanmamış veya ulaşılamayan bir hedef, bulkcopy() öğesinin Failed to retrieve destination metadata zaman aşımı hatasıyla başarısız olmasına neden olur.

Varsayılan olarak, bulkcopy() bir satırdaki her değeri sıra konumuna göre tablodaki bir sütunla eşler; bu nedenle her sütunun sırayla bir değeri olmalıdır. Tabloda bir identity sütunu varsa veya yalnızca bazı sütunları dolduruyorsanız, hedef sütunları açıkça belirtmek için column_mappings parametresini kullanın. Aksi takdirde değerler yanlış sütunlara kayır ve bulkcopy() başarısız olur.

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

İkili veri işlemleri

Microsoft SQL’in HASHBYTES işlevini, değerlerin tamamını almadan ikili içeriği karşılaştırmak için kullanın.

İkili verileri karşılaştırın

Microsoft SQL Server'da içerik hash'larını hesaplayıp karşılaştırarak aynı içeriği paylaşan ikili satırları bulun:

# 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

Python'da hash hesaplama

SHA-256 hash'larını Python'da hesaplayın ve bunları ikili verilerle birlikte saklayın:

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

İkili veri kodlama ve çözme

İkili verileri Base64 ile kodlanmış dizelere ve bu dizelerden dönüştürme:

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

Belirli ikili formatlarla çalışma

Bu örnekler, yaygın dosya formatlarının depolanmadan önce nasıl doğrulanıp ekleneceğini gösterir.

PDF belgeleri

PDF dosya imzalarını eklemeden önce doğrulayın:

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

PIL/Pillow ile Görseller

Fotoğrafları saklamadan önce Pillow (PIL) ile boyutlandırarak alan tasarrufu yapın:

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

Sıkıştırılmış veriler

Eklemeden önce ikili veri gzip ile sıkıştırarak depolamayı azaltın:

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)

En iyi uygulamalar

Bu yönergelerlikleri uygulayarak ikili veri için doğru sütun tipini ve depolama stratejisini seçin.

Uygun sütun tipleri kullanın

Verilerinizin özelliklerine göre sütun tipini seçin:

Farklı ikili kullanım durumları için, uygun Microsoft SQL veri tipini seçin:

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

Büyük dosyalar için FILESTREAM'i düşünün

Büyük dosyalar (1 MB'dan fazla) için, verileri dosya sisteminde depolayan Microsoft SQL FILESTREAM özelliğini düşünün. FILESTREAM, kullanımdan önce sunucu tarafı yapılandırması gerektirir:

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

İkili veri doğrulama

İkili veri depolamadan önce sihirli baytları (dosya imzalarını) kontrol ederek dosya tiplerini doğrulayın:

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

İlgili içerik

• Veri türü eşlemeleri
• Toplu kopya
• NULL değerleri işle