Vous trouverez ci-dessous un exemple autonome de stockage de fichiers directement dans un champ blob via Flask-Admin. La vérification des erreurs est minimale, mais cela devrait vous aider à aller dans la bonne direction.
Les parties importantes du code :
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 définit quels champs sont stockés avec les données blob, il est souvent utile de transporter des informations supplémentaires telles que la taille du fichier, le type mime et le nom du fichier téléchargé d'origine.
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 est la table de base de données qui stocke le blob (via le BlobMixin). Dans ce cas, nous donnons à chaque image un nom unique qui est indépendant du nom du fichier téléchargé.
Classe BlobUploadField(fields.StringField) est presque une copie de la classe FileUploadField de Flask-Admin
. Il existe cependant quelques différences importantes - nous devons savoir quels champs nous utilisons pour stocker la taille du fichier, le type mime et le nom du fichier d'origine. Ceux-ci sont transmis via le constructeur et sont utilisés dans le def populate_obj(self, obj, name) méthode.
Classe ImageView(ModelView) est une vue Flask-Admin simple. Notez comment le champ blob est défini dans form_extra_fields . Nous construisons un BlobUploadField et transmettre la liste des extensions de fichier autorisées, le nom de champ de taille, le nom de champ de nom de fichier et le nom de champ de type mime. Les noms des champs (taille, nom de fichier et type mime) sont tirés directement du BlobMixin noms de champs 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'
)}
J'ai ajouté une colonne de lien de téléchargement à la vue de liste avec un formateur de colonne approprié afin que vous puissiez cliquer sur le lien et télécharger le fichier.
Code ci-dessous testé avec Python 2.7.9, Flask 0.10.0, Flask-Admin 1.1.0 et Flask-SQLAlchemy 2.0. Utilise la base de données en mémoire SQLite afin que les données soient perdues lors de la fermeture de l'application 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)
