mssql-python 驅動程式支援 binary、 、 varbinary及 image Microsoft SQL 的資料型別。
Microsoft SQL 以以下欄位類型儲存二進位資料：
| 類型 | 描述 | 大小上限 |
|---|---|---|
binary(n) |
固定長度的二進位資料。 | 8,000 位元組 |
varbinary(n) |
可變長度的二進位資料。 | 8,000 位元組 |
varbinary(max) |
大型二進位資料。 | 2 GB |
image |
舊有大型二進位資料（已棄用）。 | 2 GB |
mssql-python 驅動程式會以 Python bytes 物件的形式回傳二進位資料。
何時將二進位儲存在資料庫中，何時儲存在檔案系統：
- 當檔案很小（低於 1 MB）、交易一致性與其他資料很重要，或需要同時備份資料和檔案時，就應該存放在資料庫中。
- 當檔案很大（超過 1 MB）、需要 CDN 傳送，或需要直接將檔案提供給客戶端且不需資料庫往返時，就儲存在檔案系統（或 Azure Blob 儲存體）。
- 作為中間方案，Microsoft SQL 的 FILESTREAM 功能以交易一致性將資料儲存在檔案系統中。
插入二進位資料
將 Python bytes 物件作為參數傳遞;驅動程式會將它們映射到變分欄位。
直接插入位元組
建立一個 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")
擷取二進位數據
驅動程式會將 varbinary 與 binary 欄位值作為 Python bytes 物件傳回。
擷取二進位資料欄
查詢二進位欄位並檢查返回 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}
)
Note
驅動程式通常透過 SQLDescribeParam推斷參數型別，但無法解析 暫存表格 或 表格變數的型別元資料。 在未使用 setinputsizes() 的情況下，將 None 插入 temp object 的 binary 或 varbinary 欄位會引發 ProgrammingError: Implicit conversion from data type varchar to varbinary(max) is not allowed。 依序為每個參數傳遞一個項目，並對二進位資料行使用 SQL 型別常數，例如 mssql_python.SQL_VARBINARY。 對於一般（永久）表格，驅動程式會自動解析類型，你可以直接通過 None 。
大型二進位資料
對於超過幾 MB 的檔案，請改以單次 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() 會依序數位置將一列中的每個值映射到資料表欄位，因此每欄必須有相應的值。 當表格有身份欄位或只填充部分欄位時，請明確傳入 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"]
二進位資料操作
使用 Microsoft SQL 函HASHBYTES式來比較二進位內容，而不必取得完整值。
比較二進位資料
透過計算並比較 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 計算雜湊值
用 Python 計算 SHA-256 雜湊值，並與二進位資料一同儲存：
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 MB），可以考慮 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})