Python MySQL 文件存储最佳实践：安全高效地处理二进制数据268

在许多应用程序中，需要将文件存储到数据库中，这对于需要持久化存储用户上传的图像、文档或其他二进制数据的情况尤为重要。本文将探讨如何使用 Python 和 MySQL 高效且安全地存储文件。我们将涵盖从数据库设计到错误处理的各个方面，并提供最佳实践，以确保您的应用程序的可靠性和性能。

一、数据库设计：LOB 字段的选择

MySQL 提供了两种主要的数据类型来存储大型二进制对象 (BLOB)：LONGBLOB 和 MEDIUMBLOB。选择哪种类型取决于文件的大小。LONGBLOB 可以存储高达 4GB 的数据，而 MEDIUMBLOB 的限制为 16MB。如果你的文件大小可能超过 16MB，那么 LONGBLOB 是更好的选择。在设计数据库表时，需要一个字段来存储文件的实际数据，另一个字段来存储文件元数据，例如文件名、文件类型和上传日期等。以下是一个示例表结构：```sql
CREATE TABLE files (
id INT AUTO_INCREMENT PRIMARY KEY,
filename VARCHAR(255) NOT NULL,
filetype VARCHAR(50) NOT NULL,
upload_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
filedata LONGBLOB
);
```

二、Python 代码实现：文件上传与存储

我们将使用 Python 的 `` 库连接到 MySQL 数据库。以下代码演示了如何将文件上传到数据库：```python
import
import os
def upload_file(filename, db_config):
"""Uploads a file to the MySQL database."""
try:
# 打开文件读取二进制数据
with open(filename, "rb") as file:
file_data = ()
# 连接到数据库
mydb = (db_config)
cursor = ()
# 获取文件名和文件类型
base, ext = (filename)
filetype = ext[1:].lower() # 去除点号并转为小写
# SQL 插入语句，使用参数化查询防止SQL注入
sql = "INSERT INTO files (filename, filetype, filedata) VALUES (%s, %s, %s)"
val = (filename, filetype, file_data)
(sql, val)
()
print(f"File '{filename}' uploaded successfully.")
except as err:
print(f"Database error: {err}")
()
except FileNotFoundError:
print(f"File '{filename}' not found.")
except Exception as e:
print(f"An error occurred: {e}")
finally:
if mydb.is_connected():
()
()

# 数据库配置 (替换为你的配置)
db_config = {
'user': 'your_username',
'password': 'your_password',
'host': 'your_host',
'database': 'your_database'
}
# 上传文件
upload_file("path/to/your/", db_config)
```

三、Python 代码实现：文件下载

以下代码展示如何从数据库中检索并下载文件：```python
import
def download_file(file_id, output_filename, db_config):
"""Downloads a file from the MySQL database."""
try:
mydb = (db_config)
cursor = (dictionary=True) # 使用 dictionary=True 获取字典结果
# 使用参数化查询防止SQL注入
sql = "SELECT filedata FROM files WHERE id = %s"
val = (file_id,)
(sql, val)
result = ()
if result:
file_data = result['filedata']
with open(output_filename, "wb") as file:
(file_data)
print(f"File with ID {file_id} downloaded successfully.")
else:
print(f"File with ID {file_id} not found.")
except as err:
print(f"Database error: {err}")
except Exception as e:
print(f"An error occurred: {e}")
finally:
if mydb.is_connected():
()
()
# 下载文件
download_file(1, "", db_config)
```