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을 추론하지만, 임시 테이블 이나 테이블 변수의 타입 메타데이터를 해석할 수는 없습니다.
None 없이 임시 객체의 binary 또는 varbinary 열에 setinputsizes()을 삽입하면 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() 데이터는 별도의 연결을 통해 서버에 로드되므로, 목적지 테이블은 이미 존재하고 다른 세션에 커밋되어 있어야 합니다. 같은 스크립트에서 자동 커밋을 끈 상태로 테이블을 생성하는 경우, conn.commit()보다 먼저 bulkcopy()를 호출하세요. 로컬 임시 테이블(#name)은 생성 세션에 비공개이기 때문에 지원되지 않습니다. 약속되지 않았거나 도달할 수 없는 목적지는 타임아웃과 함께 bulkcopy() 실패를 초래합니다Failed to retrieve destination metadata.
한정되지 않은 테이블 이름, schema.table 또는 database.schema.table을(를) 사용하세요. 서버로 한정된 4부분 이름은 대상 메타데이터를 검색하기 전에 거부됩니다. 자세한 내용은 대량 복사 작업을 참조하세요.
기본적으로 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})