目录

  1. 引言：理解“读取图片”的含义
    • 1.1 从数据库中读取 BLOB 的流程
    • 1.2 再次强调 BLOB 存储策略的考量
  2. Python 与 SQLite 交互基础
    • 2.1 建立数据库连接
    • 2.2 获取游标对象
    • 2.3 执行 SQL SELECT 语句
    • 2.4 获取查询结果
    • 2.5 关闭连接
  3. 从 SQLite 数据库中检索图片数据
    • 3.1 准备数据库和示例图片 (承接“插入图片”内容)
    • 3.2 构建 SELECT 查询
    • 3.3 处理检索到的二进制数据
    • Python 代码实现：检索并保存单张图片
  4. 检索多张图片或特定条件的图片
    • 4.1 检索所有图片
    • 4.2 通过条件检索图片
    • Python 代码实现：批量检索图片
  5. 最佳实践与注意事项
    • 5.1 参数化查询 (防止 SQL 注入)
    • 5.2 使用 with 语句管理连接
    • 5.3 错误处理 (try-except)
    • 5.4 文件命名与扩展名处理
    • 5.5 性能与内存管理
    • 5.6 BLOB 存储的替代方案 (路径存储)
  6. 综合代码示例
  7. 总结

1. 引言：理解“读取图片”的含义

当我们说“从 SQLite 中读取图片”时，通常是指从数据库中提取以二进制大对象（BLOB）形式存储的图像数据，并将其转换回可以在文件系统上查看的图片文件格式。这个过程是“将图片插入到 SQLite”的逆操作。

1.1 从数据库中读取 BLOB 的流程

  1. 连接到数据库：使用 sqlite3.connect() 建立与 SQLite 数据库的连接。
  2. 执行 SELECT 查询：编写 SQL SELECT 语句来选择包含图像数据的 BLOB 列，通常还会选择其他元数据（如图片名称、MIME 类型）。
  3. 获取查询结果：使用游标对象的 fetchone() 或 fetchall() 方法获取结果。图像数据将以 Python bytes 对象的D类型返回。
  4. 将 bytes 数据写入文件：使用 Python 的文件操作，以二进制写入模式 ('wb') 打开一个新文件，并将从数据库中获取的 bytes 数据写入该文件。

1.2 再次强调 BLOB 存储策略的考量

虽然本指南详细讲解了如何读取 BLOB 图片，但再次提醒：对于大多数应用场景，尤其是涉及到大量或大尺寸图片时，将图片存储在文件系统中，并在数据库中仅存储其路径 是更推荐的做法。直接存储 BLOB 会导致数据库文件膨胀，影响性能和可维护性。只有在特定需求（如单文件部署、强制事务一致性、小尺寸图标等）下，才考虑直接存储 BLOB。

2. Python 与 SQLite 交互基础

从 SQLite 中读取数据的基本步骤与写入数据类似。

2.1 建立数据库连接

import sqlite3
import os
import mimetypes # 用于猜测文件扩展名

DB_FILE = "image_gallery.db"

2.2 获取游标对象

cursor = conn.cursor()

2.3 执行 SQL SELECT 语句

cursor.execute("SELECT data, name, mimetype FROM images WHERE id = ?", (image_id,))

2.4 获取查询结果

  • cursor.fetchone()：获取查询结果的第一行数据。
  • cursor.fetchall()：获取查询结果的所有行数据。

结果中的 BLOB 数据会作为 Python 的 bytes 类型返回。

2.5 关闭连接

最好使用 with sqlite3.connect(...) as conn: 语句，它会确保连接在代码块结束后自动关闭。

3. 从 SQLite 数据库中检索图片数据

我们将以上一篇关于“插入图片”的示例为基础，假设数据库中已经存在图片。

3.1 准备数据库和示例图片

这里我们将重用之前用于插入图片的初始化和创建图片的代码，以确保我们有一个可供读取的数据库。

# --- (部分代码承接上一篇“插入图片”的示例，用于初始化数据库和插入示例图片) ---
# 为了本篇内容的独立性，我们在此再次定义这些辅助函数。
import shutil
from PIL import Image # 需要安装 Pillow: pip install Pillow

TEST_IMAGE_FOLDER = "test_images_for_read"
RETRIEVED_IMAGE_FOLDER = "retrieved_images_from_db" # 新的输出目录

def create_sample_images(num_images=3):
    """
    创建一些用于测试的假图片文件。
    """
    os.makedirs(TEST_IMAGE_FOLDER, exist_ok=True)
    print(f"Creating {num_images} sample images in '{TEST_IMAGE_FOLDER}'...")
    sample_images_paths = []
    for i in range(num_images):
        width, height = (100 + i * 10), (50 + i * 10)
        color = (255 - i * 50, 100 + i * 30, 50 + i * 70) # 不同的颜色
        img_path = os.path.join(TEST_IMAGE_FOLDER, f"sample_image_{i+1}.png")
        try:
            img = Image.new('RGB', (width, height), color=color)
            img.save(img_path)
            sample_images_paths.append(img_path)
        except ImportError:
            print("Pillow library not found. Cannot create sample images. Please install it ('pip install Pillow') or provide your own test images.")
            # Fallback for minimal png
            with open(img_path, 'wb') as f:
                 f.write(b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\x00IEND\xaeB`\x82')
            sample_images_paths.append(img_path)
    print(f"Sample images created: {sample_images_paths}")
    return sample_images_paths

def init_database():
    """初始化数据库并创建 images 表。"""
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
    
    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS images (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                data BLOB NOT NULL,
                mimetype TEXT
            );
        ''')
        conn.commit()
    print(f"Database '{DB_FILE}' initialized and 'images' table created.")

def insert_image_into_db(image_path, db_file=DB_FILE):
    """
    将指定的图片文件插入到数据库中。
    返回插入图片的 ID。
    """
    if not os.path.exists(image_path):
        print(f"Error: Image file not found at {image_path}")
        return None

    image_name = os.path.basename(image_path)
    mime_type, _ = mimetypes.guess_type(image_path)
    if mime_type is None:
        mime_type = "application/octet-stream"

    image_id = None
    try:
        with sqlite3.connect(db_file) as conn:
            cursor = conn.cursor()
            with open(image_path, 'rb') as f:
                image_data = f.read()
            
            sql = "INSERT INTO images (name, data, mimetype) VALUES (?, ?, ?);"
            cursor.execute(sql, (image_name, image_data, mime_type))
            conn.commit()
            image_id = cursor.lastrowid
            # print(f"Inserted '{image_name}' with ID: {image_id}") # 减少输出
    except sqlite3.Error as e:
        print(f"Failed to insert image '{image_name}': {e}")
    except IOError as e:
        print(f"File I/O error when reading image '{image_name}': {e}")
    return image_id

3.2 构建 SELECT 查询

最常见的查询是根据 id 来获取特定的图片。

SELECT data, name, mimetype FROM images WHERE id = ?;

3.3 处理检索到的二进制数据

  1. 从 fetchone() 返回的元组中提取 bytes 数据。
  2. 构建输出文件的完整路径和名称。可以根据存储的 mimetype 或原始 name 来推断文件扩展名。
  3. 使用 open(output_path, 'wb') 以二进制写入模式创建文件。
  4. 调用 file_object.write(image_data) 将 bytes 数据写入文件。

Python 代码实现：检索并保存单张图片

def retrieve_and_save_image(image_id, output_folder=RETRIEVED_IMAGE_FOLDER, db_file=DB_FILE):
    """
    从数据库中检索指定 ID 的图片，并保存到文件。
    :param image_id: 要检索的图片 ID。
    :param output_folder: 保存检索图片的目录。
    :param db_file: SQLite 数据库文件路径。
    :return: 保存的图片文件的完整路径，如果失败则返回 None。
    """
    os.makedirs(output_folder, exist_ok=True) # 确保输出目录存在
    
    print(f"\nRetrieving image with ID {image_id} from database...")
    saved_path = None
    try:
        with sqlite3.connect(db_file) as conn:
            cursor = conn.cursor()
            sql = "SELECT data, name, mimetype FROM images WHERE id = ?;"
            cursor.execute(sql, (image_id,))
            result = cursor.fetchone()

            if result:
                image_data, original_name, mime_type = result
                
                # 推断文件扩展名：优先使用MIME类型，其次是原始文件名，最后是默认二进制
                extension = mimetypes.guess_extension(mime_type)
                if not extension and '.' in original_name:
                    extension = os.path.splitext(original_name)[1]
                if not extension:
                    extension = '.bin' # 默认二进制扩展名

                # 构建输出文件名，为了避免冲突，通常会加上ID或时间戳
                base_name_without_ext = os.path.splitext(original_name)[0]
                output_filename = f"{base_name_without_ext}_retrieved_id{image_id}{extension}"
                output_path = os.path.join(output_folder, output_filename)
                
                with open(output_path, 'wb') as f:
                    f.write(image_data)
                print(f"Image '{original_name}' (ID: {image_id}) retrieved and saved to: {output_path}")
                saved_path = output_path
            else:
                print(f"No image found with ID: {image_id}")
    except sqlite3.Error as e:
        print(f"Database error during retrieval of image ID {image_id}: {e}")
    except IOError as e:
        print(f"File I/O error when writing retrieved image: {e}")
    return saved_path

4. 检索多张图片或特定条件的图片

4.1 检索所有图片

如果你想检索数据库中的所有图片，只需省略 WHERE 子句。

SELECT id, data, name, mimetype FROM images;

4.2 通过条件检索图片

可以使用任何列作为 WHERE 子句的条件，例如按 name 或 mimetype 检索。

SELECT id, data, name, mimetype FROM images WHERE mimetype = ?;

Python 代码实现：批量检索图片

def retrieve_all_images(output_folder=RETRIEVED_IMAGE_FOLDER, db_file=DB_FILE):
    """
    从数据库中检索所有图片，并分别保存到文件。
    """
    os.makedirs(output_folder, exist_ok=True)
    print(f"\nRetrieving all images from database to '{output_folder}'...")
    retrieved_count = 0
    try:
        with sqlite3.connect(db_file) as conn:
            cursor = conn.cursor()
            sql = "SELECT id, data, name, mimetype FROM images;"
            cursor.execute(sql)
            results = cursor.fetchall()

            if results:
                for img_id, image_data, original_name, mime_type in results:
                    extension = mimetypes.guess_extension(mime_type)
                    if not extension and '.' in original_name:
                        extension = os.path.splitext(original_name)[1]
                    if not extension:
                        extension = '.bin'

                    base_name_without_ext = os.path.splitext(original_name)[0]
                    output_filename = f"{base_name_without_ext}_retrieved_id{img_id}{extension}"
                    output_path = os.path.join(output_folder, output_filename)
                    
                    with open(output_path, 'wb') as f:
                        f.write(image_data)
                    print(f"  Saved image '{original_name}' (ID: {img_id}) to: {output_path}")
                    retrieved_count += 1
                print(f"Successfully retrieved and saved {retrieved_count} images.")
            else:
                print("No images found in the database.")
    except sqlite3.Error as e:
        print(f"Database error during batch retrieval: {e}")
    except IOError as e:
        print(f"File I/O error during batch writing: {e}")
    return retrieved_count

5. 最佳实践与注意事项

5.1 参数化查询 (防止 SQL 注入)

在构建 SELECT 语句中的 WHERE 子句时，如果条件值来自外部输入（例如用户输入的 ID 或名称），务必使用参数化查询。这对于 BLOB 数据的检索同样重要，可以有效防止 SQL 注入攻击。

# 正确示例
cursor.execute("SELECT data FROM images WHERE id = ?", (user_provided_id,))

5.2 使用 with 语句管理连接

sqlite3.connect() 对象和文件对象都是上下文管理器。始终使用 with 语句来确保数据库连接和文件句柄在操作完成后自动关闭，即使发生异常也能正确释放资源。

5.3 错误处理 (try-except)

使用 try...except sqlite3.Error 和 try...except IOError 来捕获可能发生的数据库错误和文件操作错误，并进行适当的处理，而不是让程序崩溃。

5.4 文件命名与扩展名处理

  • 唯一性：为了避免覆盖现有文件，在保存检索到的图片时，可以考虑在文件名中包含数据库 ID、时间戳或一个 UUID。
  • 扩展名：尽量利用存储在数据库中的 MIME 类型 (mimetypes.guess_extension()) 或原始文件名来确定正确的文件扩展名，这有助于系统正确识别文件类型。如果无法确定，可以默认使用 .bin 或 .dat。

5.5 性能与内存管理

  • 大尺寸 BLOB：如果图片非常大，一次性将整个 BLOB 加载到内存 (f.read()) 可能会消耗大量内存。对于极大的文件，可以考虑分块读取和写入，但这在 Python 的 sqlite3 模块中通常不是直接必要的，因为 sqlite3 在内部处理大 BLOB 已经很高效了。
  • 批量操作：在检索大量图片时，fetchall() 会将所有结果加载到内存中。如果内存是问题，可以考虑逐行处理结果（虽然 sqlite3 的 cursor 默认情况下已经倾向于缓存所有结果）。
  • 按需检索：只在需要时才从数据库中检索图片，而不是一次性检索所有图片。

5.6 BLOB 存储的替代方案 (路径存储)

再次重申，如果数据库文件大小、性能或与其他服务的集成是主要考虑因素，那么将图片存储在外部文件系统，并在数据库中只存储指向这些文件的路径，通常是更好的选择。这种情况下，“读取图片”就变成了：

  1. 从数据库中读取图片文件的路径。
  2. 使用 Python 的文件 I/O 直接从该路径读取图片文件。

6. 综合代码示例

import sqlite3
import os
import mimetypes
import shutil
import datetime
from PIL import Image # 需要安装 Pillow: pip install Pillow

# --- 配置 ---
DB_FILE = "image_gallery_full_example.db"
TEST_IMAGE_FOLDER = "test_images_for_full_example"
RETRIEVED_IMAGE_FOLDER = "retrieved_images_full_example"

# --- 辅助函数：创建测试图片 ---
def create_sample_images(num_images=3):
    """创建一些用于测试的假图片文件。"""
    os.makedirs(TEST_IMAGE_FOLDER, exist_ok=True)
    sample_images_paths = []
    for i in range(num_images):
        width, height = (100 + i * 10), (50 + i * 10)
        color = (255 - i * 50, 100 + i * 30, 50 + i * 70) 
        img_path = os.path.join(TEST_IMAGE_FOLDER, f"sample_image_{i+1}.png")
        try:
            img = Image.new('RGB', (width, height), color=color)
            img.save(img_path)
            sample_images_paths.append(img_path)
        except ImportError:
            print("Pillow not found. Creating minimal PNG fallback.")
            with open(img_path, 'wb') as f:
                 f.write(b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\x00IEND\xaeB`\x82')
            sample_images_paths.append(img_path)
    return sample_images_paths

# --- 数据库操作函数 (初始化和插入 - 为了演示读取，需要先有数据) ---
def init_database():
    """初始化数据库并创建 images 表。"""
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
    
    with sqlite3.connect(DB_FILE) as conn:
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS images (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                data BLOB NOT NULL,
                mimetype TEXT
            );
        ''')
        conn.commit()
    print(f"Database '{DB_FILE}' initialized and 'images' table created.")

def insert_image_into_db(image_path, db_file=DB_FILE):
    """将指定的图片文件插入到数据库中，返回插入图片的 ID。"""
    if not os.path.exists(image_path): return None
    image_name = os.path.basename(image_path)
    mime_type, _ = mimetypes.guess_type(image_path)
    if mime_type is None: mime_type = "application/octet-stream"
    image_id = None
    try:
        with sqlite3.connect(db_file) as conn:
            cursor = conn.cursor()
            with open(image_path, 'rb') as f: image_data = f.read()
            sql = "INSERT INTO images (name, data, mimetype) VALUES (?, ?, ?);"
            cursor.execute(sql, (image_name, image_data, mime_type))
            conn.commit()
            image_id = cursor.lastrowid
    except sqlite3.Error as e: print(f"Failed to insert image '{image_name}': {e}")
    except IOError as e: print(f"File I/O error when reading image '{image_name}': {e}")
    return image_id

# --- 核心读取函数 ---
def retrieve_and_save_single_image(image_id, output_folder=RETRIEVED_IMAGE_FOLDER, db_file=DB_FILE):
    """
    从数据库中检索指定 ID 的图片，并保存到文件。
    """
    os.makedirs(output_folder, exist_ok=True)
    print(f"\n--- Retrieving single image with ID {image_id} ---")
    saved_path = None
    try:
        with sqlite3.connect(db_file) as conn:
            cursor = conn.cursor()
            sql = "SELECT data, name, mimetype FROM images WHERE id = ?;"
            cursor.execute(sql, (image_id,))
            result = cursor.fetchone()

            if result:
                image_data, original_name, mime_type = result
                extension = mimetypes.guess_extension(mime_type)
                if not extension and '.' in original_name: extension = os.path.splitext(original_name)[1]
                if not extension: extension = '.bin' 

                base_name_without_ext = os.path.splitext(original_name)[0]
                output_filename = f"{base_name_without_ext}_retrieved_id{image_id}{extension}"
                output_path = os.path.join(output_folder, output_filename)
                
                with open(output_path, 'wb') as f:
                    f.write(image_data)
                print(f"  Successfully retrieved and saved to: {output_path}")
                saved_path = output_path
            else:
                print(f"  No image found with ID: {image_id}")
    except sqlite3.Error as e:
        print(f"  Database error during retrieval: {e}")
    except IOError as e:
        print(f"  File I/O error when writing: {e}")
    return saved_path

def retrieve_all_images_from_db(output_folder=RETRIEVED_IMAGE_FOLDER, db_file=DB_FILE):
    """
    从数据库中检索所有图片，并分别保存到文件。
    """
    os.makedirs(output_folder, exist_ok=True)
    print(f"\n--- Retrieving all images from database ---")
    retrieved_count = 0
    try:
        with sqlite3.connect(db_file) as conn:
            cursor = conn.cursor()
            sql = "SELECT id, data, name, mimetype FROM images;"
            cursor.execute(sql)
            results = cursor.fetchall()

            if results:
                for img_id, image_data, original_name, mime_type in results:
                    extension = mimetypes.guess_extension(mime_type)
                    if not extension and '.' in original_name: extension = os.path.splitext(original_name)[1]
                    if not extension: extension = '.bin'

                    base_name_without_ext = os.path.splitext(original_name)[0]
                    output_filename = f"{base_name_without_ext}_retrieved_id{img_id}{extension}"
                    output_path = os.path.join(output_folder, output_filename)
                    
                    with open(output_path, 'wb') as f:
                        f.write(image_data)
                    # print(f"  Saved image '{original_name}' (ID: {img_id}) to: {output_path}") # 减少输出
                    retrieved_count += 1
                print(f"  Successfully retrieved and saved {retrieved_count} images.")
            else:
                print("  No images found in the database.")
    except sqlite3.Error as e:
        print(f"  Database error during batch retrieval: {e}")
    except IOError as e:
        print(f"  File I/O error during batch writing: {e}")
    return retrieved_count

def list_all_images_info(db_file=DB_FILE):
    """列出数据库中所有图片的元数据。"""
    print("\n--- Listing All Images Info in Database ---")
    try:
        with sqlite3.connect(db_file) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT id, name, mimetype FROM images;")
            images_info = cursor.fetchall()
            if not images_info:
                print("  No images found in the database.")
                return []
            for img_id, name, mimetype in images_info:
                print(f"  ID: {img_id}, Name: '{name}', MIME Type: '{mimetype}'")
            return images_info
    except sqlite3.Error as e:
        print(f"Error listing images: {e}")
        return []

def cleanup():
    """清理生成的文件和数据库。"""
    print("\n--- Cleaning up generated files ---")
    if os.path.exists(DB_FILE):
        os.remove(DB_FILE)
        print(f"  Deleted database file: {DB_FILE}")
    if os.path.exists(TEST_IMAGE_FOLDER):
        shutil.rmtree(TEST_IMAGE_FOLDER)
        print(f"  Deleted test image folder: {TEST_IMAGE_FOLDER}")
    if os.path.exists(RETRIEVED_IMAGE_FOLDER):
        shutil.rmtree(RETRIEVED_IMAGE_FOLDER)
        print(f"  Deleted retrieved image folder: {RETRIEVED_IMAGE_FOLDER}")

# --- 主执行逻辑 ---
def main():
    cleanup() # 每次运行前清理
    init_database()
    sample_image_paths = create_sample_images(3)

    inserted_ids = []
    print("\n--- Inserting sample images into DB ---")
    for path in sample_image_paths:
        img_id = insert_image_into_db(path)
        if img_id:
            inserted_ids.append(img_id)
            print(f"  Inserted '{os.path.basename(path)}' with ID: {img_id}")

    list_all_images_info()

    if inserted_ids:
        # 示例 1: 检索并保存第一个插入的图片
        retrieve_and_save_single_image(inserted_ids[0])

        # 示例 2: 尝试检索一个不存在的图片
        retrieve_and_save_single_image(999)

        # 示例 3: 检索并保存所有图片
        retrieve_all_images_from_db()
    else:
        print("\nNo images were inserted, skipping retrieval tests.")

    print("\n--- Program finished ---")
    # cleanup() # 如果需要保留生成的文件以供检查，可以取消注释掉此行

if __name__ == "__main__":
    main()

7. 总结

为您详尽解析了在 Python 中使用 sqlite3 模块从 SQLite 数据库中读取（检索）图片数据的方法。

核心要点回顾:

  • SQL SELECT: 使用 SELECT 语句检索存储为 BLOB 类型的图片数据。
  • Python bytes: 数据库返回的图像数据是 bytes 对象。
  • 文件写入: 使用 open(file_path, 'wb') 以二进制写入模式将 bytes 数据保存为图片文件。
  • 安全与健壮性: 始终使用参数化查询，利用 with 语句管理资源，并实现 try-except 错误处理。
  • 文件命名: 妥善处理输出文件的命名和扩展名，确保文件的可用性和唯一性。
  • 策略选择: 再次强调，对于大多数实际应用，将大尺寸图片存储在文件系统，并在数据库中存储路径是更优的方案。直接存储 BLOB 适用于特定的小型化、事务一致性要求高的场景。

通过遵循这些指南，您将能够安全、高效地从 SQLite 数据库中读取图像数据，并将其恢复为可用的图片文件。

# python # sqlite # jvm
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

ESP32-WROOM-32E-N4R2：自带2MB PSRAM的均衡之选，轻松应对复杂应用

avatar 亚马逊云科技技术品牌专区

JSM2318 40V N 沟道高速开关 MOSFET

随着消费电子、工业小型电源、物联网设备持续向小型化、高频化、高能效迭代，同步 Buck 降压电路对 MOS 管提出三重严苛要求：极低导通损耗、高速开关特性、优异抗干扰能力。市面上通用小信号 MOS 管往往存在栅电荷高、导通电阻偏大、高温性能衰减明显等问题，极易造成电源发热、转换效率下滑、EMI 干扰超标等工程难题。深耕功率半导体赛道的。

avatar 亚马逊云科技技术品牌专区
cover

ESP32-WROOM-32E-H8：大容量Flash宽温模组，为工业物联网存储痛点而生

avatar 亚马逊云科技技术品牌专区