简介:本资源是一套基于Flet前端框架与FastAPI后端服务协同实现的文件上传系统模板,面向Python全栈初学者及轻量级Web应用开发者,解决前后端联调中文件上传、进度反馈与本地持久化保存的核心痛点。适用于文档管理、媒体库搭建、团队项目文件共享等实际场景,兼顾功能性与可扩展性。压缩包共5个文件(3个Python主程序、1个说明文本、1个操作演示GIF),总大小108KB;其中Flet前端组件负责多文件选择与实时进度展示,FastAPI后端处理异步接收与安全存储,配套read.txt提供关键配置说明与环境变量设置指引。已有102人学习下载,附带可视化交互演示动图与模块化代码结构,开箱即用,便于快速理解前后端通信机制、文件流处理逻辑及Flet自定义UI组件封装方法。
1. 项目缘起:为什么需要一个Flet文件上传组件?
最近在做一个内部工具,需要让用户上传一些图片和文档。项目用的是Flet,这个框架挺有意思,它让你用Python就能写前端,后端逻辑也包圆了,对于全栈Python开发者来说,开发效率确实高。但当我真正开始做文件上传功能时,发现事情没那么简单。
Flet官方有FilePicker控件,用起来是方便,点一下就能选文件。但问题来了,它选完文件后,文件数据是留在前端的,你得自己写逻辑把文件数据发到后端,再保存到服务器磁盘上。官方文档的例子比较基础,真要放到项目里,你得考虑一堆事:上传进度怎么显示?文件大小要不要限制?上传前能不能预览一下图片?上传失败了怎么友好地提示用户?更别提如果你想做一个拖拽上传的优雅界面,或者一次选多个文件还要分批次上传,光靠基础的FilePicker就得写不少胶水代码。
所以,我就想,为什么不把这些通用逻辑封装成一个“开箱即用”的自定义组件呢?这个组件应该像一块乐高积木,我以后在任何Flet项目里,只要把它拿过来,设置一下保存路径、允许的文件类型,界面和功能就都有了,后端接收和保存文件的脏活累活它也全帮我干了。这就是我做这个“Flet前端上传文件后端接受保存文件自定义组件模板”的初衷——把一次性的开发,变成可复用的资产。
2. 组件核心设计:前后端职责与通信机制
在Flet里做全栈功能,首先要理清前后端的边界。Flet应用默认运行在“客户端”和“服务器端”两个部分。虽然都是Python,但“前端”的代码(你的UI控件和事件处理)会在用户的浏览器或客户端APP里执行,而“后端”的代码(你的业务逻辑、数据库操作、文件IO)则在服务器上执行。它们之间通过Flet内置的协议进行通信。
2.1 前端:不只是FilePicker的简单包装
前端组件的核心自然是ft.FilePicker,但我们要把它包装得更易用、功能更完整。
基础文件选择:这是起点。我们通过FilePicker的on_result事件来获取用户选择的文件。这里有个关键点,FilePicker选择文件后,返回的FilePickerFile对象包含了文件名、路径(在客户端是虚拟路径)和文件大小,但并不包含文件的实际字节数据。文件数据需要另外读取。
读取与准备文件数据:为了把文件发到后端,我们需要把文件转换成可以传输的格式。通常,我们会将文件读取为字节(bytes),然后进行Base64编码。Base64是一种将二进制数据编码成ASCII字符串的方法,这样就能通过JSON等文本协议安全地传输了。前端代码需要完成这个读取和编码的工作。
用户界面与交互:一个友好的上传组件不能只有一个按钮。我设计的组件前端部分包括:
- 一个触发文件选择器的按钮或区域:可以是按钮,也可以是一个支持拖拽的
Container。 - 文件列表预览区:用户选择文件后,立即显示文件名、大小和状态(等待中、上传中、成功、失败)。
- 上传进度条:对于大文件,一个动态增长的进度条是安抚用户焦虑的良药。
- 操作按钮:如“开始上传”、“取消上传”、“清空列表”等。
2.2 后端:可靠地接收与持久化
后端是真正干活的地方,它需要提供一个接口(Flet里叫page的方法或事件处理函数)来接收前端发来的文件数据。
接收Base64数据:前端会把文件名和对应的Base64字符串一起发过来。后端需要解码这个Base64字符串,还原出原始的字节数据。
安全地保存文件:这是后端最重要的职责。直接保存有风险,我们需要考虑:
- 文件名处理:用户上传的文件名可能包含特殊字符、中文,甚至是恶意路径(如
../../../etc/passwd)。我们必须对文件名进行“净化”(Sanitize),通常做法是只保留字母、数字、点、下划线和连字符,或者干脆用UUID生成一个唯一的文件名,但保留原始扩展名。 - 目录创建:确保目标保存目录存在,如果不存在就创建它。
- 写入文件:将解码后的字节数据写入到目标路径。这里要用二进制模式(
‘wb’)打开文件。
响应与错误处理:文件保存成功后,后端需要通知前端,比如返回一个成功的消息和文件在服务器上的最终保存路径(或访问URL)。如果保存失败(如磁盘已满、权限不足),也必须捕获异常,并返回明确的错误信息给前端,让用户知道发生了什么。
2.3 前后端通信:自定义事件与数据封装
Flet提供了page.pubsub.send和page.pubsub.subscribe的发布-订阅机制,非常适合这种前后端异步通信。但对于我们这个组件,我选择了一种更直接、结构更清晰的方式:自定义方法调用。
我在自定义组件类里定义了一个upload方法。前端触发上传时,实际上是调用了这个upload方法,并将文件列表数据作为参数传递进去。在这个upload方法内部,它负责组织数据(如添加时间戳、会话ID),然后调用一个在后端定义好的、专门处理文件上传的函数。
这个后端处理函数,我把它设计成可配置的。当用户使用这个组件时,他需要自己实现这个保存文件的逻辑(比如保存到项目的uploads文件夹),然后把他的这个函数“注入”到组件里。这样,组件就只负责通用的流程控制(触发、传输、状态更新),而具体的存储策略(存到本地、还是云存储)则由使用者决定,非常灵活。
数据包的结构我设计成字典格式,大概长这样:
{ “files”: [ { “name”: “example.jpg“, “data”: “/9j/4AAQSkZJRgABAQEAYABgAAD...“, # 很长的Base64字符串 “size”: 102456, “type”: “image/jpeg” } ], “upload_id”: “unique_session_id_123“ # 用于追踪一次上传会话 }3. 手把手实现:从零构建可复用的FileUploader组件
理论讲完了,我们来看代码。我将这个组件命名为FileUploader,它是一个ft.UserControl的子类。
3.1 组件前端UI构建
首先,我们初始化所有需要的控件,并把它们排列好。
import flet as ft import os, uuid, base64, asyncio from typing import Callable, List, Optional class FileUploader(ft.UserControl): def __init__(self, on_upload_complete: Optional[Callable] = None, upload_endpoint: Optional[Callable] = None, allowed_extensions: List[str] = None, max_size_mb: int = 10): super().__init__() self.on_upload_complete = on_upload_complete # 上传完成后的回调 self.upload_endpoint = upload_endpoint # 后端保存文件的函数 self.allowed_extensions = allowed_extensions or [“.jpg“, “.jpeg“, “.png“, “.pdf“, “.txt“] # 允许的文件类型 self.max_size_bytes = max_size_mb * 1024 * 1024 # 最大文件大小(字节) # 文件选择器 self.file_picker = ft.FilePicker(on_result=self._on_file_pick_result) self.selected_files = [] # 存储待上传的文件信息 # UI控件 self.upload_button = ft.ElevatedButton(“选择文件“, icon=ft.icons.UPLOAD_FILE, on_click=lambda _: self.file_picker.pick_files(allow_multiple=True, allowed_extensions=self.allowed_extensions)) self.drag_target = ft.Container( content=ft.Column([ ft.Icon(ft.icons.CLOUD_UPLOAD, size=50), ft.Text(“拖拽文件到此处或点击上方按钮“, size=16) ], alignment=ft.MainAxisAlignment.CENTER, horizontal_alignment=ft.CrossAxisAlignment.CENTER), border=ft.border.all(2, ft.colors.OUTLINE), border_radius=10, padding=30, on_hover=self._on_drag_hover, # 注意:Flet原生不支持`on_drop`,这里需要借助`GestureDetector`和自定义逻辑模拟,为简化示例,此处先聚焦点击上传。 ) self.file_list_view = ft.ListView(expand=True, spacing=10) self.progress_bar = ft.ProgressBar(width=400, visible=False) self.status_text = ft.Text(““) self.start_upload_btn = ft.ElevatedButton(“开始上传“, on_click=self._start_upload, disabled=True) self.clear_btn = ft.OutlinedButton(“清空列表“, on_click=self._clear_list) def build(self): # 将FilePicker添加到page的overlay,这是一个特殊位置 self.page.overlay.append(self.file_picker) self.page.update() return ft.Column([ self.upload_button, ft.Divider(), self.drag_target, ft.Text(“已选文件:“, weight=ft.FontWeight.BOLD), ft.Container(self.file_list_view, height=200, border=ft.border.all(1), border_radius=5, padding=10), ft.Row([self.start_upload_btn, self.clear_btn], alignment=ft.MainAxisAlignment.START), self.progress_bar, self.status_text ])_on_file_pick_result方法是关键,它处理文件选择后的逻辑:验证文件大小、类型,然后将文件信息添加到列表并更新UI。
def _on_file_pick_result(self, e: ft.FilePickerResultEvent): if not e.files: return for file in e.files: # 1. 验证文件大小 if file.size > self.max_size_bytes: self.status_text.value = f“文件 ‘{file.name}‘ 超过 {self.max_size_bytes//(1024*1024)}MB 限制“ self.status_text.color = ft.colors.RED self.update() continue # 2. 验证文件类型(通过扩展名) _, ext = os.path.splitext(file.name) if self.allowed_extensions and ext.lower() not in self.allowed_extensions: self.status_text.value = f“文件 ‘{file.name}‘ 类型不允许,仅支持 {‘, ‘.join(self.allowed_extensions)}“ self.status_text.color = ft.colors.ORANGE self.update() continue # 3. 添加到待上传列表 file_info = { “client_path“: file.path, # 注意:这是客户端路径,不能直接用于服务器 “name“: file.name, “size“: file.size, “status“: “pending“, # pending, reading, uploading, success, error “progress“: 0 } self.selected_files.append(file_info) # 4. 更新UI列表项 self._add_file_to_list_view(file_info) self.start_upload_btn.disabled = len(self.selected_files) == 0 self.update()3.2 核心难点:从前端读取文件数据并传输
这是整个组件最核心也最容易出问题的一步。在Web环境中,Flet运行在浏览器里,我们可以用FilePicker的get_files方法吗?不行,这个方法只在某些特定模式下可用。更通用的做法是,我们需要在前端(JavaScript)读取文件内容。
但是,我们的组件是纯Python的UserControl,怎么执行JavaScript呢?这里就需要用到Flet的js模块,它允许我们直接调用前端的JavaScript代码。我们可以在前端定义一个JavaScript函数来读取文件为Base64。
不过,为了简化示例和保证逻辑清晰,我采用一种更“Flet”的方式:将文件读取和编码的逻辑,也放在一个后端函数里,但通过前端的FilePicker事件来触发。听起来有点绕,具体是这样:
- 用户点击
FilePicker,选择文件(这是前端操作)。 FilePicker的on_result事件触发,我们在事件处理函数里,获得了文件的引用(file.path)。- 我们不在前端读取文件,而是立刻将这个文件的引用(路径)发送到后端。
- 后端收到这个路径后,在服务器端环境,根据这个路径去读取文件内容,并编码为Base64。
等一下!这里有个巨大的陷阱:file.path是客户端文件系统的路径(比如C:\Users\...\xxx.jpg或/User/.../xxx.jpg),这个路径在服务器端是无效的!服务器根本访问不到用户电脑上的C:\盘。
所以,这条路走不通。正确的做法是:文件数据必须在客户端(浏览器)读取,然后传输到服务器。我们必须借助Flet与前端JavaScript交互的能力。由于这部分涉及较深的Flet与JS交互,且代码较长,我将其核心思路概括为:
- 在Flet控件初始化时,向页面注入一段自定义的JavaScript代码,这段代码包含一个全局函数,例如
window.readFileAsBase64。 - 当
FilePicker选择文件后,我们通过page.invoke_method调用一个自定义的“后端方法”,并将文件的path传递过去。 - 在这个“后端方法”里,我们并不直接读文件,而是再次通过
page.eval_js,去调用我们之前注入的window.readFileAsBase64函数,并传入文件的path。 - JavaScript函数在浏览器环境中执行,读取该路径对应的文件(浏览器有安全限制,只能读取用户主动选择的文件),将其转换为Base64字符串。
- JavaScript函数通过
window.flet.send将Base64字符串发送回Flet的后端Python代码。 - 后端Python代码收到Base64数据,此时才真正开始我们之前设计的保存流程。
这个过程是异步的,涉及多次前后端往返。为了简化示例代码的复杂度,我在这里先提供一个模拟版本的核心上传函数_start_upload,它假设我们已经通过某种方式(比如上述的JS交互)将selected_files中的client_path替换成了真实的base64_data。
async def _start_upload(self, e): if not self.upload_endpoint: self.status_text.value = “错误:未配置上传处理函数(upload_endpoint)“ self.status_text.color = ft.colors.RED self.update() return self.start_upload_btn.disabled = True self.progress_bar.visible = True self.status_text.value = “开始上传...“ self.status_text.color = ft.colors.BLUE self.update() total_files = len(self.selected_files) for index, file_info in enumerate(self.selected_files): # 更新状态为上传中 file_info[“status“] = “uploading“ self._update_file_list_item(file_info) try: # !!! 这里是关键 !!! # 假设 file_info 中已经包含了 ‘base64_data‘ 字段,这是通过前端JS读取后添加的。 # 在实际完整实现中,需要在上一步(_on_file_pick_result)之后,触发一个异步操作去获取base64_data。 if “base64_data“ not in file_info: # 模拟获取数据的过程(实际应调用JS) await asyncio.sleep(0.5) # 模拟网络延迟 # 这里应该是一段调用page.eval_js并等待返回的复杂逻辑 # 为演示,我们假设数据已就绪 file_info[“base64_data“] = “SIMULATED_BASE64_DATA_FOR_” + file_info[“name“] # 准备上传数据包 upload_payload = { “file_name“: file_info[“name“], “file_data“: file_info[“base64_data“], # 这里是完整的Base64字符串 “file_size“: file_info[“size“], “upload_id“: str(uuid.uuid4())[:8] # 生成一个简短的上传会话ID } # 调用用户提供的后端处理函数 # 这个函数应该接收 upload_payload, 解码并保存文件,然后返回结果字典 result = await self.upload_endpoint(upload_payload) if result.get(“success“): file_info[“status“] = “success“ file_info[“saved_path“] = result.get(“saved_path“, ““) else: file_info[“status“] = “error“ file_info[“error“] = result.get(“message“, “未知错误“) file_info[“progress“] = 100 except Exception as ex: file_info[“status“] = “error“ file_info[“error“] = str(ex) file_info[“progress“] = 0 finally: # 更新单个文件的UI self._update_file_list_item(file_info) # 更新总进度条 self.progress_bar.value = (index + 1) / total_files self.update() # 所有文件处理完毕 self.progress_bar.visible = False self.status_text.value = f“上传完成!成功 {len([f for f in self.selected_files if f[‘status‘]==‘success‘])} 个,失败 {len([f for f in self.selected_files if f[‘status‘]==‘error‘])} 个。“ self.status_text.color = ft.colors.GREEN self.start_upload_btn.disabled = False self.update() # 触发上传完成回调 if self.on_upload_complete: completed_files = [f for f in self.selected_files if f[‘status‘] in [‘success‘, ‘error‘]] self.on_upload_complete(completed_files)3.3 后端保存函数示例
现在,我们来看看使用者需要提供的upload_endpoint函数应该长什么样。这是一个标准的后端处理函数,它运行在服务器端。
import os, base64, uuid from pathlib import Path async def handle_file_upload(upload_data: dict) -> dict: “““ 处理文件上传的后端函数。 参数 upload_data: 包含 ‘file_name‘, ‘file_data‘ (base64), ‘file_size‘, ‘upload_id‘ 的字典。 返回: 包含 ‘success‘ (bool) 和 ‘message‘ 或 ‘saved_path‘ 的字典。 “““ file_name = upload_data.get(“file_name“) file_data_b64 = upload_data.get(“file_data“) upload_id = upload_data.get(“upload_id“, ““) if not file_name or not file_data_b64: return {“success“: False, “message“: “文件名或数据为空“} try: # 1. 安全处理文件名 # 提取扩展名 _, ext = os.path.splitext(file_name) ext = ext.lower() # 生成一个安全的唯一文件名,保留原扩展名 safe_filename = f“{uuid.uuid4().hex}{ext}“ # 或者,如果你想保留原文件名但净化它: # import re # safe_filename = re.sub(r‘[^\w\.-]‘, ‘_‘, file_name) # 2. 定义保存目录(示例:当前目录下的 ‘uploads‘ 文件夹,按日期分目录) from datetime import datetime today_str = datetime.now().strftime(“%Y-%m-%d“) save_dir = Path(f“./uploads/{today_str}“) save_dir.mkdir(parents=True, exist_ok=True) # 递归创建目录 # 3. 构建完整保存路径 save_path = save_dir / safe_filename # 4. 解码Base64并写入文件 # Base64数据可能包含头信息(如`data:image/png;base64,`),需要剥离 if ‘,‘ in file_data_b64: # 去掉类似 `data:image/png;base64,` 的前缀 file_data_b64 = file_data_b64.split(‘,‘)[1] file_bytes = base64.b64decode(file_data_b64) with open(save_path, ‘wb‘) as f: f.write(file_bytes) # 5. 返回成功信息 # 可以返回相对路径或可用于访问的URL return { “success“: True, “message“: “文件保存成功“, “saved_path“: str(save_path), # 或者返回一个URL路径,如 f“/uploads/{today_str}/{safe_filename}“ “original_name“: file_name, “saved_name“: safe_filename } except base64.binascii.Error: return {“success“: False, “message“: “Base64数据解码失败“} except IOError as e: return {“success“: False, “message“: f“文件写入失败: {str(e)}“} except Exception as e: return {“success“: False, “message“: f“处理过程中发生未知错误: {str(e)}“}4. 在Flet应用中集成与使用
组件和后台函数都准备好了,现在看看怎么在Flet的main函数里把它们拼装起来。
import flet as ft def main(page: ft.Page): page.title = “Flet文件上传组件演示“ page.theme_mode = ft.ThemeMode.LIGHT page.padding = 30 # 定义一个回调函数,在上传完成后被调用 def on_upload_finished(file_list): print(f“上传完成回调,共处理 {len(file_list)} 个文件。“) for f in file_list: status = “成功“ if f[‘status‘]==‘success‘ else “失败“ print(f“ - {f[‘name‘]}: {status}“) if f[‘status‘]==‘success‘: print(f“ 保存位置: {f.get(‘saved_path‘, ‘N/A‘)}“) # 创建上传组件实例 # 参数说明: # on_upload_complete: 所有文件处理完后的回调 # upload_endpoint: 后端处理函数 # allowed_extensions: 允许上传的文件扩展名 # max_size_mb: 单个文件最大大小(MB) uploader = FileUploader( on_upload_complete=on_upload_finished, upload_endpoint=handle_file_upload, # 这里传入我们写好的后端函数 allowed_extensions=[“.jpg“, “.jpeg“, “.png“, “.gif“, “.pdf“, “.docx“], max_size_mb=5 ) # 将组件添加到页面 page.add( ft.Text(“文件上传演示“, size=24, weight=ft.FontWeight.BOLD), ft.Divider(), uploader ) # 启动应用 ft.app(target=main)运行这个程序,你会看到一个包含文件选择按钮、文件列表和操作按钮的界面。点击“选择文件”,选取几个符合要求的文件,它们会出现在列表中。点击“开始上传”,组件就会开始模拟上传过程(因为我们还没有实现真实的前端Base64读取,所以用的是模拟数据),并调用你写的handle_file_upload函数。如果一切正常,你会在项目目录下看到一个uploads/年-月-日/的文件夹,里面保存着上传的文件。
5. 避坑指南与进阶优化
在实际使用和开发这个组件的过程中,我踩过不少坑,也想到一些可以优化的地方。
坑1:文件大小验证的时机。我们是在前端选择文件后立刻验证大小的,这依赖于FilePickerFile.size属性。但请注意,这个size属性在Web环境中是可靠的,因为它来自浏览器的File API。然而,为了绝对安全,在后端保存文件之前,应该再次检查写入前数据流的大小,防止恶意客户端伪造小文件信息却发送大文件内容。
坑2:Base64传输的性能与内存。Base64编码会让文件体积增大约33%。对于超大文件(比如几百MB的视频),将整个文件读入内存并编码成Base64字符串,可能会导致浏览器内存不足(OOM)。对于大文件,更好的方案是使用分块上传(Chunked Upload)。前端将文件切成小块(如1MB一块),分别读取、编码、上传,后端则按顺序接收并拼接这些块。这需要更复杂的前后端协议设计,但能显著提升大文件上传的可靠性和用户体验。
坑3:文件名冲突与安全。我们用了UUID来生成文件名,这解决了冲突问题。但有时业务需要保留原始文件名。这时,净化文件名就至关重要。不能简单替换非法字符,因为不同操作系统的非法字符集不同。一个相对安全的做法是使用werkzeug.utils.secure_filename(如果你安装了Flask/Werkzeug)或实现类似的逻辑,它会把非ASCII字符、空格、特殊符号都处理掉。
进阶优化1:实现真实的拖拽上传。Flet的Container有on_hover和on_click,但没有原生的on_drop。要实现拖拽,需要用到ft.GestureDetector,并监听其on_pan_update等事件,通过计算鼠标位置和移动轨迹来模拟拖放效果,或者更直接地,使用ft.Markdown或ft.Html控件嵌入一个带有HTML5拖拽API的迷你前端,这涉及到Flet与前端更深的混合编程。
进阶优化2:上传暂停与继续。在网络不稳定的环境下,这个功能非常有用。实现思路是:在前端记录每个文件块的上传状态,当暂停时,停止发送请求;当继续时,从中断的块开始发送。后端需要支持接收非连续的文件块,并能正确地将它们按偏移量写入文件的指定位置。
进阶优化3:与云存储集成。现在的upload_endpoint是把文件存到本地服务器磁盘。你可以轻松地修改它,将文件字节数据直接上传到阿里云OSS、腾讯云COS或AWS S3。只需要将handle_file_upload函数中的本地文件写入逻辑,替换成对应云服务商SDK的上传方法即可。这样,你的Flet应用就拥有了直接上传到云存储的能力。
把这个组件模板搭建起来后,我发现它确实大大简化了后续项目中所有涉及文件上传功能的开发。虽然第一次实现时,前后端数据交换的部分花了些时间琢磨,但一旦封装好,它就变成了一个可靠的“黑盒”。下次再需要上传功能,我只需要关心两件事:允许上传什么文件,以及文件最终要存到哪里。至于中间那些繁琐的交互、验证和传输逻辑,就全部交给这个FileUploader组件了。
本文还有配套的精品资源,点击获取