我正在开发一个需要上传图片的后端应用程序,我正在使用烧瓶上传,但我有几个问题,要么给我文件夹 = f"user_{user_id}" 的语法错误 ^ SyntaxError: invalid syntax
也适用于正则表达式 = f"^[a-zA-Z0-9][a-zA-Z0-9_()-.]*.({allowed_format})$"
请帮助我如何解决它,并让我知道如何将图像路径 url 存储到用户的数据库表中。
图像帮助代码:
import os
import re
from typing import Union
from werkzeug.datastructures import FileStorage
from flask_uploads import UploadSet, IMAGES
IMAGE_SET = UploadSet("images", IMAGES) # set name and allowed extensions
def save_image(image, folder, name):
return IMAGE_SET.save(image, folder, name)
def get_path(filename, folder):
return IMAGE_SET.path(filename, folder)
def find_image_any_format(filename, folder):
# """
# Given a format-less filename, try to find the file by appending each of the allowed formats to the given
# filename and check if the file exists
# :param filename: formatless filename
# :param folder: the relative folder in which to search
# :return: the path of the image if exists, otherwise None
# """
for _format in IMAGES: # look for existing avatar and delete it
image = f"{filename}.{_format}"
image_path = IMAGE_SET.path(filename=image, folder=folder)
if os.path.isfile(image_path):
return image_path
return None
def _retrieve_filename(file):
"""
Make our filename related functions generic, able to deal with FileStorage object as well as filename str.
"""
if isinstance(file, FileStorage):
return file.filename
return file
def is_filename_safe(file):
"""
Check if a filename is secure according to our definition
- starts with a-z A-Z 0-9 at least one time
- only contains a-z A-Z 0-9 and _().-
- followed by a dot (.) and a allowed_format at the end
"""
filename = _retrieve_filename(file)
allowed_format = "|".join(IMAGES)
# format IMAGES into regex, eg: ('jpeg','png') --> 'jpeg|png'
regex = f"^[a-zA-Z0-9][a-zA-Z0-9_()-\.]*\.({allowed_format})$"
return re.match(regex, filename) is not None
def get_basename(file):
"""
Return file's basename, for example
get_basename('some/folder/image.jpg') returns 'image.jpg'
"""
filename = _retrieve_filename(file)
return os.path.split(filename)[1]
def get_extension(file):
"""
Return file's extension, for example
get_extension('image.jpg') returns '.jpg'
"""
filename = _retrieve_filename(file)
return os.path.splitext(filename)[1]
图片资源代码:
from flask_restful import Resource
from flask_uploads import UploadNotAllowed
from flask import send_file, request
from flask_jwt_extended import jwt_required, get_jwt_identity
import traceback
import os
from libs import image_helper
from schemas.image import ImageSchema
image_schema = ImageSchema()
class ImageUpload(Resource):
@jwt_required
def post(self):
# """
# This endpoint is used to upload an image file. It uses the
# JWT to retrieve user information and save the image in the user's folder.
# If a file with the same name exists in the user's folder, name conflicts
# will be automatically resolved by appending a underscore and a smallest
# unused integer. (eg. filename.png to filename_1.png).
# """
data = image_schema.load(request.files)
user_id = get_jwt_identity()
folder = f"user_{user_id}"
try:
# save(self, storage, folder=None, name=None)
image_path = image_helper.save_image(data[0], folder=folder)
# here we only return the basename of the image and hide the internal folder structure from our user
basename = image_helper.get_basename(image_path)
return {"message": ("image_uploaded").format(basename)}, 201
except UploadNotAllowed: # forbidden file type
extension = image_helper.get_extension(data["image"])
return {"message": ("image_illegal_extension").format(extension)}, 400
class Image(Resource):
@jwt_required
def get(self, filename):
# """
# This endpoint returns the requested image if exists. It will use JWT to
# retrieve user information and look for the image inside the user's folder.
# """
user_id = get_jwt_identity()
folder = f"user_{user_id}"
# check if filename is URL secure
if not image_helper.is_filename_safe(filename):
return {"message": ("image_illegal_file_name").format(filename)}, 400
try:
# try to send the requested file to the user with status code 200
return send_file(image_helper.get_path(filename, folder=folder))
except FileNotFoundError:
return {"message": ("image_not_found").format(filename)}, 404
@jwt_required
def delete(self, filename):
# """
# This endpoint is used to delete the requested image under the user's folder.
# It uses the JWT to retrieve user information.
# """
user_id = get_jwt_identity()
folder = f"user_{user_id}"
# check if filename is URL secure
if not image_helper.is_filename_safe(filename):
return {"message": ("image_illegal_file_name").format(filename)}, 400
try:
os.remove(image_helper.get_path(filename, folder=folder))
return {"message": ("image_deleted").format(filename)}, 200
except FileNotFoundError:
return {"message": ("image_not_found").format(filename)}, 404
except:
traceback.print_exc()
return {"message": ("image_delete_failed")}, 500
图像架构:
from marshmallow import Schema, fields
from werkzeug.datastructures import FileStorage
# from werkzeug.utils import secure_filename
# from werkzeug.datastructures import FileStorage
class FileStorageField(fields.Field):
default_error_messages = {
"invalid": "Not a valid image."
}
def _deserialize(self, value, attr, data):
if value is None:
return None
if not isinstance(value, FileStorage):
self.fail("invalid")
return value
class ImageSchema(Schema):
image = FileStorageField(required=True)