Abaixo está um exemplo independente de armazenamento de arquivos diretamente em um campo blob via Flask-Admin. Há uma verificação mínima de erros, mas deve levá-lo na direção certa.
As partes importantes do código:
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
class define quais campos são armazenados junto com os dados do blob, geralmente é útil carregar informações adicionais, como tamanho do arquivo, tipo mime e o nome do arquivo original carregado. 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
class é a tabela de banco de dados que armazena o blob (por meio do BlobMixin). Neste caso, estamos dando a cada imagem um nome exclusivo que é independente do nome do arquivo carregado. Classe
BlobUploadField(fields.StringField)
é quase uma cópia da classe FileUploadField de Flask-Admin
. No entanto, existem algumas diferenças importantes - precisamos saber quais campos estamos usando para armazenar o tamanho do arquivo, o tipo mime e o nome do arquivo original. Estes são passados através do construtor e são usados no def populate_obj(self, obj, name)
método. Classe
ImageView(ModelView)
é uma visualização direta do Flask-Admin. Observe como o campo blob é definido em form_extra_fields
. Estamos construindo um BlobUploadField
e passando a lista de extensões de arquivo permitidas, o nome do campo de tamanho, o nome do campo do nome do arquivo e o nome do campo do tipo mime. Os nomes dos campos (tamanho, nome do arquivo e tipo MIME) são obtidos diretamente do BlobMixin
nomes de campos de classe. 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'
)}
Adicionei uma coluna de link de download à exibição de lista com um formatador de coluna apropriado para que você possa clicar no link e baixar o arquivo.
Código abaixo testado usando Python 2.7.9, Flask 0.10.0, Flask-Admin 1.1.0 e Flask-SQLAlchemy 2.0. Usa o banco de dados na memória SQLite para que os dados sejam perdidos quando o aplicativo do frasco for fechado.
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)