简介本资源是一套基于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_resultself._on_file_pick_result) self.selected_files [] # 存储待上传的文件信息 # UI控件 self.upload_button ft.ElevatedButton(“选择文件“, iconft.icons.UPLOAD_FILE, on_clicklambda _: self.file_picker.pick_files(allow_multipleTrue, allowed_extensionsself.allowed_extensions)) self.drag_target ft.Container( contentft.Column([ ft.Icon(ft.icons.CLOUD_UPLOAD, size50), ft.Text(“拖拽文件到此处或点击上方按钮“, size16) ], alignmentft.MainAxisAlignment.CENTER, horizontal_alignmentft.CrossAxisAlignment.CENTER), borderft.border.all(2, ft.colors.OUTLINE), border_radius10, padding30, on_hoverself._on_drag_hover, # 注意Flet原生不支持on_drop这里需要借助GestureDetector和自定义逻辑模拟为简化示例此处先聚焦点击上传。 ) self.file_list_view ft.ListView(expandTrue, spacing10) self.progress_bar ft.ProgressBar(width400, visibleFalse) self.status_text ft.Text(““) self.start_upload_btn ft.ElevatedButton(“开始上传“, on_clickself._start_upload, disabledTrue) self.clear_btn ft.OutlinedButton(“清空列表“, on_clickself._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(“已选文件“, weightft.FontWeight.BOLD), ft.Container(self.file_list_view, height200, borderft.border.all(1), border_radius5, padding10), ft.Row([self.start_upload_btn, self.clear_btn], alignmentft.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(parentsTrue, exist_okTrue) # 递归创建目录 # 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_completeon_upload_finished, upload_endpointhandle_file_upload, # 这里传入我们写好的后端函数 allowed_extensions[“.jpg“ “.jpeg“ “.png“ “.gif“ “.pdf“ “.docx“], max_size_mb5 ) # 将组件添加到页面 page.add( ft.Text(“文件上传演示“ size24 weightft.FontWeight.BOLD), ft.Divider(), uploader ) # 启动应用 ft.app(targetmain)运行这个程序你会看到一个包含文件选择按钮、文件列表和操作按钮的界面。点击“选择文件”选取几个符合要求的文件它们会出现在列表中。点击“开始上传”组件就会开始模拟上传过程因为我们还没有实现真实的前端Base64读取所以用的是模拟数据并调用你写的handle_file_upload函数。如果一切正常你会在项目目录下看到一个uploads/年-月-日/的文件夹里面保存着上传的文件。5. 避坑指南与进阶优化在实际使用和开发这个组件的过程中我踩过不少坑也想到一些可以优化的地方。坑1文件大小验证的时机。我们是在前端选择文件后立刻验证大小的这依赖于FilePickerFile.size属性。但请注意这个size属性在Web环境中是可靠的因为它来自浏览器的File API。然而为了绝对安全在后端保存文件之前应该再次检查写入前数据流的大小防止恶意客户端伪造小文件信息却发送大文件内容。坑2Base64传输的性能与内存。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组件了。本文还有配套的精品资源点击获取 Flet文件上传组件开发指南：从原理到实践
简介本资源是一套基于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_resultself._on_file_pick_result) self.selected_files [] # 存储待上传的文件信息 # UI控件 self.upload_button ft.ElevatedButton(“选择文件“, iconft.icons.UPLOAD_FILE, on_clicklambda _: self.file_picker.pick_files(allow_multipleTrue, allowed_extensionsself.allowed_extensions)) self.drag_target ft.Container( contentft.Column([ ft.Icon(ft.icons.CLOUD_UPLOAD, size50), ft.Text(“拖拽文件到此处或点击上方按钮“, size16) ], alignmentft.MainAxisAlignment.CENTER, horizontal_alignmentft.CrossAxisAlignment.CENTER), borderft.border.all(2, ft.colors.OUTLINE), border_radius10, padding30, on_hoverself._on_drag_hover, # 注意Flet原生不支持on_drop这里需要借助GestureDetector和自定义逻辑模拟为简化示例此处先聚焦点击上传。 ) self.file_list_view ft.ListView(expandTrue, spacing10) self.progress_bar ft.ProgressBar(width400, visibleFalse) self.status_text ft.Text(““) self.start_upload_btn ft.ElevatedButton(“开始上传“, on_clickself._start_upload, disabledTrue) self.clear_btn ft.OutlinedButton(“清空列表“, on_clickself._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(“已选文件“, weightft.FontWeight.BOLD), ft.Container(self.file_list_view, height200, borderft.border.all(1), border_radius5, padding10), ft.Row([self.start_upload_btn, self.clear_btn], alignmentft.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(parentsTrue, exist_okTrue) # 递归创建目录 # 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_completeon_upload_finished, upload_endpointhandle_file_upload, # 这里传入我们写好的后端函数 allowed_extensions[“.jpg“ “.jpeg“ “.png“ “.gif“ “.pdf“ “.docx“], max_size_mb5 ) # 将组件添加到页面 page.add( ft.Text(“文件上传演示“ size24 weightft.FontWeight.BOLD), ft.Divider(), uploader ) # 启动应用 ft.app(targetmain)运行这个程序你会看到一个包含文件选择按钮、文件列表和操作按钮的界面。点击“选择文件”选取几个符合要求的文件它们会出现在列表中。点击“开始上传”组件就会开始模拟上传过程因为我们还没有实现真实的前端Base64读取所以用的是模拟数据并调用你写的handle_file_upload函数。如果一切正常你会在项目目录下看到一个uploads/年-月-日/的文件夹里面保存着上传的文件。5. 避坑指南与进阶优化在实际使用和开发这个组件的过程中我踩过不少坑也想到一些可以优化的地方。坑1文件大小验证的时机。我们是在前端选择文件后立刻验证大小的这依赖于FilePickerFile.size属性。但请注意这个size属性在Web环境中是可靠的因为它来自浏览器的File API。然而为了绝对安全在后端保存文件之前应该再次检查写入前数据流的大小防止恶意客户端伪造小文件信息却发送大文件内容。坑2Base64传输的性能与内存。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组件了。本文还有配套的精品资源点击获取