Работа с бинарными данными

Драйвер mssql-python поддерживает binary, varbinary, и image типы данных для Microsoft SQL.

Microsoft SQL хранит бинарные данные в следующих столбцах:

Type	Описание	Максимальный размер
binary(n)	Двоичные данные фиксированной длины.	8 000 байт
varbinary(n)	Двоичные данные с переменной длиной.	8 000 байт
varbinary(max)	Большие двоичные данные.	2 ГБ
image	Устаревшие двоичные данные большого объёма (не рекомендуется к использованию).	2 ГБ

Драйвер mssql-python возвращает бинарные данные в виде объектов Pythonbytes.

Когда хранить бинарный файл в базе данных по сравнению с файловой системой:

• Храните в базе данных, когда файлы маленькие (менее 1 МБ), важна согласованность транзакций с другими данными, или нужно делать резервные копии данных и файлов вместе.
• Храните файлы в файловой системе (или в Хранилище BLOB-объектов Azure), если файлы большие (более 1 МБ), вам нужна доставка через CDN или если нужно отдавать файлы клиентам напрямую без дополнительных обращений к базе данных.
• В качестве компромисса функция FILESTREAM от Microsoft SQL хранит данные в файловой системе с транзакционной согласованностью.

Вставить бинарные данные

Передавайте объекты Python bytes в качестве параметров; драйвер сопоставляет их со столбцами типа varbinary.

Вставлять байты напрямую

Создайте объект Python bytes и передайте его как параметр запроса:

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

Вставка из файла

Прочитайте файл в двоичном режиме и вставьте его содержимое:

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

Вставить файл изображения

Сохраняйте файлы изображений с метаданными, обнаруживая типы MIME по расширениям файлов:

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

Получение двоичных данных

Драйвер возвращает значения варбинарных и бинарных столбцов в виде объектов Pythonbytes.

Извлечение двоичного столбца

Запросите бинарные столбцы и проверьте возвращённый 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

Сохранить в файл

Извлечь двоичные данные и записать их на диск:

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

Экспорт бинарных строк в файлы

Экспорт нескольких двоичных строк в локальные файлы:

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

Передайте None, чтобы вставить значение SQL NULL в столбец с двоичными данными. Поскольку #Documents является временной таблицей, сначала объявите типы параметров с помощью setinputsizes(), чтобы драйвер привязал Content как 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}
)

Замечание

Драйвер обычно выводит типы параметров через SQLDescribeParam, но не может разрешить метаданные типа для временной таблицы или переменной таблицы. Вставка None в столбец binary или varbinary временного объекта без setinputsizes() вызывает ProgrammingError: Implicit conversion from data type varchar to varbinary(max) is not allowed. Передайте по одной записи на параметр по порядку и используйте константу типа SQL, например mssql_python.SQL_VARBINARY , для двоичного столбца. Для обычной (постоянной) таблицы драйвер автоматически определяет тип, и вы можете передать None напрямую.

Большие двоичные данные

Для файлов объёмом на несколько мегабайт вставляйте всё содержимое в одну запись varbinary(max ) вместо того, чтобы читать файл небольшими частями.

Передавайте большие файлы потоком

Для больших файлов обрабатывайте частями:

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

Используйте массовое копирование для двоичных данных

Используйте этот bulkcopy() метод для эффективной вставки нескольких бинарных файлов за одну операцию.

Important

bulkcopy() загружает данные по отдельному соединению с сервером, поэтому таблица назначения должна уже существовать, быть коммитированной и видимой для других сессий. Если вы создаёте таблицу в том же скрипте с выключенным автокоммитом, вызовите conn.commit() перед bulkcopy(). Локальные временные таблицы (#name) не поддерживаются, потому что они приватны для сессии, которая их создала. Отсутствие обязательств или недостижимая цель приводит bulkcopy() к провалу с Failed to retrieve destination metadata тайм-аутом.

По умолчанию bulkcopy() сопоставляет каждое значение в строке со столбцом таблицы по порядковому номеру, поэтому значения должны быть заданы для всех столбцов и в правильном порядке. Если в таблице есть столбец IDENTITY или вы заполняете только некоторые столбцы, передайте column_mappings, чтобы явно указать имена целевых столбцов. В противном случае значения смещаются на неправильные столбцы и bulkcopy() не работают.

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

Операции с бинарными данными

Используйте функцию HASHBYTES Microsoft SQL для сравнения двоичного контента без получения полных значений.

Сравните бинарные данные

Найдите двоичные строки, разделяющие одинаковый контент, вычисляя и сравнивая хэши в 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

Вычисление хэша в Python

Вычислите хэши SHA-256 в Python и храните их вместе с бинарными данными:

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

Кодировать и декодировать бинарные данные

Преобразование бинарных данных в строки, закодированные в 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")

Работа с определёнными двоичными форматами

Эти примеры показывают, как проверить и вставить распространённые форматы файлов перед их хранением.

PDF-документы

Проверьте подписи PDF-файлов перед их вставкой:

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

Изменяйте размер изображений с помощью Pillow (PIL) перед их хранением, чтобы сэкономить место:

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

Сжатые данные

Сократите хранение, сжав бинарные данные с помощью gzip перед вставкой:

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)

Лучшие практики

Применяйте эти рекомендации для выбора правильного типа столбца и стратегии хранения двоичных данных.

Используйте соответствующие типы столбцов

Выберите тип столбца в зависимости от характеристик ваших данных:

Для различных бинарных сценариев использования выберите подходящий тип данных 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

Рассмотрим FILESTREAM для больших файлов

Для больших файлов (более 1 МБ) рассмотрим функцию Microsoft SQL FILESTREAM, которая хранит данные в файловой системе. FILESTREAM требует настройки на стороне сервера перед использованием:

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

Проверка бинарных данных

Проверьте типы файлов, проверяя магические байты (сигнатуры файлов) перед хранением двоичных данных:

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

Связанные материалы

• Сопоставления типов данных
• Массовое копирование
• Обработка значений NULL