Skip to main content
 首页 » 编程设计

python之使用 Flask-admin 在数据库中存储 PDF 文件

2025年05月04日12davidwang456

我可以使用 Flask-Admin 将 PDF 文件作为对象或 blob 存储在数据库中吗? 我没有在文档中找到任何引用。

谢谢。 干杯

请您参考如下方法:

下面是一个自包含的示例,通过 Flask-Admin 将文件直接存储在 blob 字段中。错误检查最少,但它应该让您朝着正确的方向前进。

代码的重要部分:

class BlobMixin(object): 
    mimetype = db.Column(db.Unicode(length=255), nullable=False) 
    filename = db.Column(db.Unicode(length=255), nullable=False) 
    blob = db.Column(db.LargeBinary(), nullable=False) 
    size = db.Column(db.Integer, nullable=False) 

BlobMixin 类定义了哪些字段与 blob 数据一起存储,通常它用于携带附加信息,例如文件大小、mime 类型和原始上传文件的文件名。

class Image(db.Model, BlobMixin): 
    __tablename__ = 'images' 
 
    id = db.Column(db.Integer, primary_key=True) 
    name = db.Column(db.Unicode(length=255), nullable=False, unique=True) 
 
    def __unicode__(self): 
        return u"name : {name}; filename : {filename})".format(name=self.name, filename=self.filename) 

Image 类是存储 blob 的数据库表(通过 BlobMixin)。在这种情况下,我们为每张图片提供了一个独立于上传文件名的唯一名称。

类 BlobUploadField(fields.StringField) 几乎是 Flask-Admin 中 FileUploadField 类的副本.不过有一些重要的区别——我们需要知道我们使用哪些字段来存储文件大小、mime 类型和原始文件名。它们通过构造函数传入,并在 def populate_obj(self, obj, name) 方法中使用。

类 ImageView(ModelView) 是一个简单的 Flask-Admin View 。请注意 blob 字段是如何在 form_extra_fields 中定义的。我们正在构造一个 BlobUploadField 并传入允许的文件扩展名列表、大小字段名、文件名字段名和 mime 类型字段名。字段名称(大小、文件名和 mimetype)直接取自 BlobMixin 类字段名称。

form_extra_fields = {'blob': BlobUploadField( 
    label='File', 
    allowed_extensions=['pdf', 'doc', 'docx', 'xls', 'xlsx', 'png', 'jpg', 'jpeg', 'gif'], 
    size_field='size', 
    filename_field='filename', 
    mimetype_field='mimetype' 
)} 

我已经使用适当的列格式器向 ListView 添加了一个下载链接列,因此您可以单击该链接并下载文件。

下面的代码使用 Python 2.7.9、Flask 0.10.0、Flask-Admin 1.1.0 和 Flask-SQLAlchemy 2.0 进行了测试。使用 SQLite 内存数据库,因此在关闭 Flask 应用程序时数据将丢失。

import io 
from gettext import gettext 
from flask import Flask, send_file 
from flask.ext.admin import Admin 
from flask.ext.admin.contrib.sqla import ModelView 
from flask.ext.sqlalchemy import SQLAlchemy 
from markupsafe import Markup 
from werkzeug.datastructures import FileStorage 
from wtforms import ValidationError, fields 
from wtforms.validators import required 
from wtforms.widgets import HTMLString, html_params, FileInput 
 
try: 
    from wtforms.fields.core import _unset_value as unset_value 
except ImportError: 
    from wtforms.utils import unset_value 
 
app = Flask(__name__) 
app.config['DEBUG'] = True 
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:' 
app.config['SQLALCHEMY_ECHO'] = True 
db = SQLAlchemy(app) 
 
 
def build_db(): 
    db.drop_all() 
    db.create_all() 
 
 
class BlobMixin(object): 
    mimetype = db.Column(db.Unicode(length=255), nullable=False) 
    filename = db.Column(db.Unicode(length=255), nullable=False) 
    blob = db.Column(db.LargeBinary(), nullable=False) 
    size = db.Column(db.Integer, nullable=False) 
 
 
class Image(db.Model, BlobMixin): 
    __tablename__ = 'images' 
 
    id = db.Column(db.Integer, primary_key=True) 
    name = db.Column(db.Unicode(length=255), nullable=False, unique=True) 
 
    def __unicode__(self): 
        return u"name : {name}; filename : {filename})".format(name=self.name, filename=self.filename) 
 
 
class BlobUploadField(fields.StringField): 
 
    widget = FileInput() 
 
    def __init__(self, label=None, allowed_extensions=None, size_field=None, filename_field=None, mimetype_field=None, **kwargs): 
 
        self.allowed_extensions = allowed_extensions 
        self.size_field = size_field 
        self.filename_field = filename_field 
        self.mimetype_field = mimetype_field 
        validators = [required()] 
 
        super(BlobUploadField, self).__init__(label, validators, **kwargs) 
 
    def is_file_allowed(self, filename): 
        """ 
            Check if file extension is allowed. 
 
            :param filename: 
                File name to check 
        """ 
        if not self.allowed_extensions: 
            return True 
 
        return ('.' in filename and 
                filename.rsplit('.', 1)[1].lower() in 
                map(lambda x: x.lower(), self.allowed_extensions)) 
 
    def _is_uploaded_file(self, data): 
        return (data and isinstance(data, FileStorage) and data.filename) 
 
    def pre_validate(self, form): 
        super(BlobUploadField, self).pre_validate(form) 
        if self._is_uploaded_file(self.data) and not self.is_file_allowed(self.data.filename): 
            raise ValidationError(gettext('Invalid file extension')) 
 
    def process_formdata(self, valuelist): 
        if valuelist: 
            data = valuelist[0] 
            self.data = data 
 
    def populate_obj(self, obj, name): 
 
        if self._is_uploaded_file(self.data): 
 
            _blob = self.data.read() 
 
            setattr(obj, name, _blob) 
 
            if self.size_field: 
                setattr(obj, self.size_field, len(_blob)) 
 
            if self.filename_field: 
                setattr(obj, self.filename_field, self.data.filename) 
 
            if self.mimetype_field: 
                setattr(obj, self.mimetype_field, self.data.content_type) 
 
 
class ImageView(ModelView): 
 
    column_list = ('name', 'size', 'filename', 'mimetype', 'download') 
    form_columns = ('name', 'blob') 
 
    form_extra_fields = {'blob': BlobUploadField( 
        label='File', 
        allowed_extensions=['pdf', 'doc', 'docx', 'xls', 'xlsx', 'png', 'jpg', 'jpeg', 'gif'], 
        size_field='size', 
        filename_field='filename', 
        mimetype_field='mimetype' 
    )} 
 
    def _download_formatter(self, context, model, name): 
        return Markup("<a href='{url}' target='_blank'>Download</a>".format(url=self.get_url('download_blob', id=model.id))) 
 
    column_formatters = { 
        'download': _download_formatter, 
    } 
 
 
# download route 
 
@app.route("/download/<int:id>", methods=['GET']) 
def download_blob(id): 
    _image = Image.query.get_or_404(id) 
    return send_file( 
        io.BytesIO(_image.blob), 
        attachment_filename=_image.filename, 
        mimetype=_image.mimetype 
    ) 
 
# Create admin 
admin = Admin(app, name='Admin', url='/') 
admin.add_view(ImageView(model=Image, session=db.session, category='Database', name='Images')) 
 
 
@app.before_first_request 
def first_request(): 
    build_db() 
 
if __name__ == '__main__': 
    app.run(debug=True, port=7777)