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

Question

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

谢谢。干杯

Answer 1

下面是一个自包含的示例，通过 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)