mssql-python 드라이버는 Microsoft SQL용 , , binaryvarbinary 데이터 타입을 지원합니다image.
Microsoft SQL은 다음과 같은 열 유형으로 이진 데이터를 저장합니다:
| Type | 설명 | 최대 크기 |
|---|---|---|
binary(n) |
고정 길이 이진 데이터. | 8,000바이트 |
varbinary(n) |
가변 길이 이진 데이터입니다. | 8,000바이트 |
varbinary(max) |
큰 이진 데이터. | 2GB |
image |
레거시 대형 이진 데이터 (폐기됨). | 2GB |
mssql-python 드라이버는 이진 데이터를 Python bytes 객체로 반환합니다.
데이터베이스에 바이너리를 저장할 때와 파일 시스템에 저장할 때:
- 파일 크기가 작거나(1MB 미만), 다른 데이터와의 거래 일관성이 중요하거나, 데이터와 파일을 함께 백업해야 할 때는 데이터베이스에 저장하세요.
- 파일이 1MB 이상, CDN 전달이 필요하거나 데이터베이스 왕복 없이 클라이언트에게 직접 파일을 제공해야 할 때는 파일 시스템(또는 Azure Blob Storage)에 저장하세요.
- 중간 지점으로는 Microsoft SQL의 FILESTREAM 기능이 트랜잭션 내 일관성을 가지고 파일 시스템에 데이터를 저장합니다.
이진 데이터 삽입
Python bytes 객체를 매개변수로 전달하고; 드라이버가 이를 varbinary column에 매핑합니다.
바이트를 직접 삽입하기
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 이진 값을 삽입하세요
이진 열에 SQL NULL을 삽입하려면 None를 전달합니다.
#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을 추론하지만, 임시 테이블 이나 테이블 변수의 타입 메타데이터를 해석할 수는 없습니다.
setinputsizes() 없이 임시 객체의 binary 또는 varbinary 열에 None을 삽입하면 ProgrammingError: Implicit conversion from data type varchar to varbinary(max) is not allowed가 발생합니다. 각 매개변수마다 한 번씩 순서대로 전달하고, 이진 열과 같은 mssql_python.SQL_VARBINARY SQL 타입 상수를 사용하세요. 일반(영구) 테이블의 경우, 드라이버가 타입을 자동으로 해결해서 바로 패스 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() 한 번의 작업으로 여러 개의 바이너리 파일을 효율적으로 삽입할 수 있습니다.
중요합니다
bulkcopy() 데이터는 별도의 연결을 통해 서버에 로드되므로, 목적지 테이블은 이미 존재하고 다른 세션에 커밋되어 있어야 합니다. 같은 스크립트에서 자동 커밋을 끈 상태로 테이블을 생성하는 경우, bulkcopy()보다 먼저 conn.commit()를 호출하세요. 로컬 임시 테이블(#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을 고려해 보세요
1MB 이상의 큰 파일의 경우, 파일 시스템에 데이터를 저장하는 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})