A continuación se muestra un ejemplo autónomo de almacenamiento de archivos directamente en un campo de blob a través de Flask-Admin. La verificación de errores es mínima, pero debería llevarlo en la dirección correcta.
Las partes importantes del 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 qué campos se almacenan junto con los datos del blob; a menudo, es útil llevar información adicional, como el tamaño del archivo, el tipo mime y el nombre del archivo original cargado.
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 es la tabla de la base de datos que almacena el blob (a través de BlobMixin). En este caso, le damos a cada imagen un nombre único que es independiente del nombre del archivo cargado.
Clase BlobUploadField(fields.StringField) es casi una copia de la clase FileUploadField de Flask-Admin
. Sin embargo, existen algunas diferencias importantes:necesitamos saber qué campos estamos usando para almacenar el tamaño del archivo, el tipo mime y el nombre del archivo original. Estos se pasan a través del constructor y se usan en def populate_obj(self, obj, name) método.
Clase ImageView(ModelView) es una vista sencilla de Flask-Admin. Tenga en cuenta cómo se define el campo blob en form_extra_fields . Estamos construyendo un BlobUploadField y pasando la lista de extensiones de archivo permitidas, el nombre de campo de tamaño, el nombre de campo de nombre de archivo y el nombre de campo de tipo MIME. Los nombres de los campos (tamaño, nombre de archivo y tipo MIME) se toman directamente del BlobMixin nombres de campo de clase.
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'
)}
Agregué una columna de enlace de descarga a la vista de lista con un formateador de columnas adecuado para que pueda hacer clic en el enlace y descargar el archivo.
El siguiente código se probó con Python 2.7.9, Flask 0.10.0, Flask-Admin 1.1.0 y Flask-SQLAlchemy 2.0. Utiliza la base de datos en memoria SQLite, por lo que los datos se perderán cuando se cierre la aplicación del matraz.
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)
