Flet 高级指南

官方文档：https://docs.flet.dev/
适用版本：Flet 0.82+（2026-05-08 核实）

目录

1. Flet高级指南 · 项目架构与文件结构
2. Flet高级指南 · 全局状态管理
3. Flet高级指南 · 高级自定义控件
4. Flet高级指南 · 性能优化
5. Flet高级指南 · PubSub 多会话通信
6. Flet高级指南 · 与第三方库集成
7. Flet高级指南 · 认证与安全
8. Flet高级指南 · 打包与部署
9. Flet高级指南 · 从零构建复杂应用

项目架构与文件结构

推荐目录结构

my_app/
├── pyproject.toml          # 项目配置与依赖
├── src/
│   ├── main.py             # 入口文件
│   ├── state/
│   │   ├── __init__.py
│   │   ├── app_store.py    # 全局状态
│   │   └── models.py       # 数据模型
│   ├── components/
│   │   ├── __init__.py
│   │   ├── navbar.py       # 导航栏组件
│   │   ├── sidebar.py      # 侧边栏组件
│   │   └── common.py       # 通用组件
│   ├── pages/
│   │   ├── __init__.py
│   │   ├── home.py         # 首页
│   │   ├── detail.py       # 详情页
│   │   └── settings.py     # 设置页
│   ├── services/
│   │   ├── __init__.py
│   │   ├── api.py          # HTTP 请求封装
│   │   └── storage.py      # 本地存储封装
│   └── assets/
│       ├── icon.png
│       └── splash.png

pyproject.toml 完整配置

[project]
name = "my_flet_app"
version = "1.0.0"
requires-python = ">=3.11"
dependencies = [
    "flet>=0.82",
    "httpx>=0.25",
    "pydantic>=2.0",
]

[tool.flet]
org = "com.example"
product = "My App"
company = "My Company"
copyright = "Copyright © 2026"

[tool.flet.app]
path = "src"
module = "main"

[tool.flet.splash]
color = "#1a1a2e"
dark_color = "#1a1a2e"

main.py 入口结构

import flet as ft
from state.app_store import AppStore
from pages.home import HomePage
from pages.detail import DetailPage
from pages.settings import SettingsPage
import re


def main(page: ft.Page):
    page.title = "My App"
    page.theme_mode = ft.ThemeMode.SYSTEM
    page.padding = 0

    store = AppStore()

    @ft.component
    def Router() -> ft.Control:
        route = page.route or "/"

        if route == "/":
            return HomePage(store=store)

        match = re.match(r"^/item/(\d+)$", route)
        if match:
            item_id = int(match.group(1))
            return DetailPage(store=store, item_id=item_id)

        if route == "/settings":
            return SettingsPage(store=store)

        return ft.Text("404 - 页面未找到", size=24)

    def on_route_change(e: ft.RouteChangeEvent):
        page.render(Router)

    page.on_route_change = on_route_change
    page.render(Router)


ft.app(main)

全局状态管理

Store 模式

将全局状态集中到一个 Store 对象，所有组件通过 Store 读写状态，避免 prop drilling（属性层层传递）。

# state/models.py
from dataclasses import dataclass, field
import flet as ft


@ft.observable
@dataclass
class User:
    id: int = 0
    name: str = ""
    email: str = ""
    is_authenticated: bool = False


@ft.observable
@dataclass
class Task:
    id: int = 0
    title: str = ""
    done: bool = False
    created_at: str = ""

# state/app_store.py
from dataclasses import dataclass, field
from typing import Optional
import flet as ft
from .models import User, Task


@ft.observable
@dataclass
class AppStore:
    # 用户状态
    current_user: User = field(default_factory=User)

    # 任务列表（使用 ObservableList）
    tasks: list[Task] = field(default_factory=list)

    # UI 状态
    is_loading: bool = False
    error_message: str = ""
    theme_mode: str = "system"

    # --- 用户操作 ---

    def set_user(self, user_data: dict):
        self.current_user.id = user_data["id"]
        self.current_user.name = user_data["name"]
        self.current_user.email = user_data["email"]
        self.current_user.is_authenticated = True

    def logout(self):
        self.current_user.is_authenticated = False
        self.current_user.id = 0
        self.current_user.name = ""

    # --- 任务操作 ---

    def add_task(self, title: str):
        import time
        task = Task(
            id=int(time.time() * 1000),
            title=title,
            done=False,
            created_at=time.strftime("%Y-%m-%d %H:%M"),
        )
        self.tasks = self.tasks + [task]

    def toggle_task(self, task_id: int):
        self.tasks = [
            Task(**{**t.__dict__, "done": not t.done}) if t.id == task_id else t
            for t in self.tasks
        ]

    def delete_task(self, task_id: int):
        self.tasks = [t for t in self.tasks if t.id != task_id]

    def set_loading(self, value: bool):
        self.is_loading = value

    def set_error(self, message: str):
        self.error_message = message

在组件中使用 Store

Store 作为参数传入组件，组件直接读取 store 的属性，状态变化时自动重新渲染。

# pages/home.py
import flet as ft
from state.app_store import AppStore


@ft.component
def HomePage(store: AppStore) -> ft.Control:
    new_title, set_title = ft.use_state("")

    def handle_add(e):
        if new_title.strip():
            store.add_task(new_title.strip())
            set_title("")

    task_items = [
        TaskItem(task=task, store=store)
        for task in store.tasks
    ]

    return ft.Column(
        controls=[
            ft.Text(
                f"你好，{store.current_user.name or '访客'}",
                size=24,
                weight=ft.FontWeight.BOLD,
            ),
            ft.Row(
                controls=[
                    ft.TextField(
                        value=new_title,
                        hint_text="添加新任务...",
                        expand=True,
                        on_change=lambda e: set_title(e.control.value),
                        on_submit=handle_add,
                    ),
                    ft.ElevatedButton("添加", on_click=handle_add),
                ],
            ),
            ft.Divider(),
            ft.Column(controls=task_items) if task_items else ft.Text(
                "暂无任务",
                color=ft.Colors.GREY_400,
                italic=True,
            ),
        ],
        expand=True,
        scroll=ft.ScrollMode.AUTO,
        spacing=12,
    )


@ft.component
def TaskItem(task, store: AppStore) -> ft.Control:
    return ft.ListTile(
        leading=ft.Checkbox(
            value=task.done,
            on_change=lambda e: store.toggle_task(task.id),
        ),
        title=ft.Text(
            task.title,
            style=ft.TextStyle(
                decoration=ft.TextDecoration.LINE_THROUGH if task.done else None,
                color=ft.Colors.GREY_400 if task.done else None,
            ),
        ),
        subtitle=ft.Text(task.created_at, size=12, color=ft.Colors.GREY),
        trailing=ft.IconButton(
            icon=ft.Icons.DELETE_OUTLINE,
            icon_color=ft.Colors.RED_300,
            on_click=lambda e: store.delete_task(task.id),
        ),
    )

高级自定义控件

@ft.control 类控件

@ft.control 用于创建有自己内部状态和生命周期的可复用控件，适合封装复杂的、带有副作用的 UI 单元。

import flet as ft
from dataclasses import field


@ft.control
class CountdownTimer(ft.Row):
    """倒计时控件，挂载时开始计时，卸载时停止。"""

    total_seconds: int = 60
    on_finish: callable = None

    def init(self):
        self._remaining = self.total_seconds
        self._task = None
        self._label = ft.Text(
            self._format_time(self._remaining),
            size=32,
            weight=ft.FontWeight.BOLD,
            color=ft.Colors.GREEN_400,
        )
        self._stop_btn = ft.IconButton(
            icon=ft.Icons.STOP,
            on_click=self._stop,
        )
        self.controls = [self._label, self._stop_btn]

    def _format_time(self, seconds: int) -> str:
        m, s = divmod(seconds, 60)
        return f"{m:02d}:{s:02d}"

    async def did_mount(self):
        import asyncio
        self._task = asyncio.create_task(self._tick())

    async def will_unmount(self):
        if self._task:
            self._task.cancel()

    async def _tick(self):
        import asyncio
        while self._remaining > 0:
            await asyncio.sleep(1)
            self._remaining -= 1
            color = ft.Colors.RED_400 if self._remaining <= 10 else ft.Colors.GREEN_400
            self._label.value = self._format_time(self._remaining)
            self._label.color = color
            self.update()
        if self.on_finish:
            self.on_finish()

    def _stop(self, e):
        if self._task:
            self._task.cancel()
        self._remaining = 0
        self._label.value = "已停止"
        self.update()

生命周期方法说明

带动画的进度圈控件

@ft.control
class AnimatedProgress(ft.Stack):
    """带百分比文字的动画进度圈。"""

    value: float = 0.0      # 0.0 ~ 1.0
    size: int = 100
    color: str = ft.Colors.BLUE_400

    def init(self):
        self._ring = ft.ProgressRing(
            value=self.value,
            width=self.size,
            height=self.size,
            color=self.color,
            stroke_width=self.size * 0.1,
            animate=ft.animation.Animation(500, ft.AnimationCurve.EASE_OUT),
        )
        self._label = ft.Text(
            f"{int(self.value * 100)}%",
            size=self.size * 0.22,
            weight=ft.FontWeight.BOLD,
        )
        self.controls = [
            self._ring,
            ft.Container(
                content=self._label,
                width=self.size,
                height=self.size,
                alignment=ft.alignment.center,
            ),
        ]

    def before_update(self):
        self._ring.value = self.value
        self._label.value = f"{int(self.value * 100)}%"

使用方式：

@ft.component
def ProgressDemo() -> ft.Control:
    progress, set_progress = ft.use_state(0.0)

    return ft.Column([
        AnimatedProgress(value=progress, size=120, color=ft.Colors.TEAL),
        ft.Slider(
            value=progress,
            min=0,
            max=1,
            on_change=lambda e: set_progress(e.control.value),
        ),
    ], horizontal_alignment=ft.CrossAxisAlignment.CENTER)

is_isolated 性能隔离

在 @ft.control 类中重写 is_isolated() 返回 True，该控件将独立更新，不随父控件重渲染。

@ft.control
class HeavyChart(ft.Container):
    data: list = field(default_factory=list)

    def is_isolated(self):
        return True  # 父组件更新时，此控件不重新渲染

    def init(self):
        # 渲染复杂图表
        ...

性能优化

大列表优化

使用 ft.ListView 代替 ft.Column 渲染大量数据，ListView 只渲染可见区域。

import flet as ft
from state.app_store import AppStore


@ft.component
def VirtualList(items: list) -> ft.Control:
    return ft.ListView(
        controls=[ItemCard(item=item) for item in items],
        item_extent=72,          # 固定每项高度，性能最佳
        expand=True,
        spacing=4,
    )


@ft.component
def ItemCard(item) -> ft.Control:
    return ft.ListTile(
        leading=ft.CircleAvatar(content=ft.Text(item["name"][0])),
        title=ft.Text(item["name"]),
        subtitle=ft.Text(item["description"]),
    )

GridView 大网格

@ft.component
def ImageGrid(images: list[str]) -> ft.Control:
    return ft.GridView(
        controls=[
            ft.Image(src=url, fit=ft.ImageFit.COVER, border_radius=8)
            for url in images
        ],
        max_extent=180,          # 每格最大宽度，自动计算列数
        child_aspect_ratio=1.0,  # 正方形
        run_spacing=8,
        spacing=8,
        expand=True,
    )

批量更新（异步加载）

当一次性加载大量数据时，分批次更新，避免单次 WebSocket 消息过大。

import flet as ft
import asyncio


@ft.component
def BatchLoader(load_items: callable) -> ft.Control:
    items, set_items = ft.use_state([])
    loading, set_loading = ft.use_state(False)

    async def load_all():
        set_loading(True)
        all_data = await load_items()
        batch_size = 50
        loaded = []
        for i in range(0, len(all_data), batch_size):
            batch = all_data[i:i + batch_size]
            loaded = loaded + batch
            set_items(list(loaded))
            await asyncio.sleep(0)  # 让出事件循环，避免阻塞 UI
        set_loading(False)

    # 组件挂载后自动开始加载（用 use_state 模拟 did_mount）
    started, set_started = ft.use_state(False)
    if not started:
        set_started(True)
        asyncio.create_task(load_all())

    if loading and not items:
        return ft.Column(
            [ft.ProgressRing(), ft.Text("加载中...")],
            horizontal_alignment=ft.CrossAxisAlignment.CENTER,
        )

    return ft.ListView(
        controls=[ft.ListTile(title=ft.Text(str(item))) for item in items],
        expand=True,
    )

避免不必要的重渲染

# 错误示范：每次渲染都创建新的 lambda，导致子控件重建
@ft.component
def BadParent(store) -> ft.Control:
    return ft.Column([
        ChildWidget(on_click=lambda e: store.do_something())  # 每次都是新函数
    ])

# 正确做法：将回调定义在 store 或模型方法上
@ft.component
def GoodParent(store) -> ft.Control:
    return ft.Column([
        ChildWidget(on_click=store.do_something)  # 稳定的函数引用
    ])

PubSub 多会话通信

PubSub 允许在多个用户会话（浏览器标签/连接）之间广播消息，适合聊天室、实时协作、看板等场景。

核心 API

实时聊天室示例

import flet as ft
from dataclasses import dataclass


@ft.observable
@dataclass
class ChatState:
    messages: list[dict] = None
    username: str = ""

    def __post_init__(self):
        if self.messages is None:
            self.messages = []


@ft.component
def ChatApp(page: ft.Page) -> ft.Control:
    state, _ = ft.use_state(ChatState())

    def on_message(msg: dict):
        state.messages = state.messages + [msg]

    # 订阅消息频道（组件首次渲染时执行一次）
    subscribed, set_subscribed = ft.use_state(False)
    if not subscribed:
        set_subscribed(True)
        page.pubsub.subscribe_topic("chat", on_message)

    def send_message(e):
        if state.username and (text := e.control.value.strip()):
            msg = {"user": state.username, "text": text}
            page.pubsub.send_all_on_topic("chat", msg)
            e.control.value = ""
            e.control.update()

    return ft.Column(
        controls=[
            ft.TextField(
                label="用户名",
                value=state.username,
                on_change=lambda e: setattr(state, "username", e.control.value),
            ),
            ft.Divider(),
            ft.ListView(
                controls=[
                    ft.Text(f"{m['user']}: {m['text']}")
                    for m in state.messages
                ],
                expand=True,
                auto_scroll=True,
            ),
            ft.TextField(
                hint_text="输入消息，按 Enter 发送",
                on_submit=send_message,
            ),
        ],
        expand=True,
        spacing=8,
    )


def main(page: ft.Page):
    page.title = "聊天室"

    @ft.component
    def App() -> ft.Control:
        return ChatApp(page=page)

    # 页面关闭时取消订阅
    def on_disconnect(e):
        page.pubsub.unsubscribe_all()

    page.on_disconnect = on_disconnect
    page.render(App)


ft.app(main, view=ft.AppView.WEB_BROWSER)

PubSub 使用注意事项

• PubSub 是进程内广播，不支持跨进程（多台服务器需要用 Redis 实现）
• 在会话断开时必须调用 unsubscribe_all()，否则会内存泄漏
• handler 在接收消息的线程中执行，会自动触发 observable 状态更新

与第三方库集成

httpx 异步 HTTP 请求

# services/api.py
import httpx
from typing import Any


class ApiClient:
    def __init__(self, base_url: str, token: str = ""):
        self.base_url = base_url
        self._headers = {"Authorization": f"Bearer {token}"} if token else {}

    async def get(self, path: str, **kwargs) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.base_url}{path}",
                headers=self._headers,
                **kwargs,
            )
            resp.raise_for_status()
            return resp.json()

    async def post(self, path: str, data: dict) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base_url}{path}",
                json=data,
                headers=self._headers,
            )
            resp.raise_for_status()
            return resp.json()

# 在组件中使用
import flet as ft
from services.api import ApiClient
import asyncio


@ft.observable
@dataclass
class PostsState:
    posts: list = None
    loading: bool = False
    error: str = ""

    def __post_init__(self):
        if self.posts is None:
            self.posts = []


@ft.component
def PostsList() -> ft.Control:
    state, _ = ft.use_state(PostsState())
    api = ApiClient("https://jsonplaceholder.typicode.com")

    async def fetch_posts():
        state.loading = True
        state.error = ""
        try:
            data = await api.get("/posts")
            state.posts = data[:20]  # 只取前 20 条
        except httpx.HTTPError as e:
            state.error = f"请求失败: {e}"
        finally:
            state.loading = False

    loaded, set_loaded = ft.use_state(False)
    if not loaded:
        set_loaded(True)
        asyncio.create_task(fetch_posts())

    if state.loading:
        return ft.Column(
            [ft.ProgressRing(), ft.Text("加载中...")],
            horizontal_alignment=ft.CrossAxisAlignment.CENTER,
        )

    if state.error:
        return ft.Column([
            ft.Text(state.error, color=ft.Colors.RED_400),
            ft.ElevatedButton("重试", on_click=lambda e: asyncio.create_task(fetch_posts())),
        ])

    return ft.ListView(
        controls=[
            ft.ListTile(
                title=ft.Text(post["title"], weight=ft.FontWeight.W_500),
                subtitle=ft.Text(post["body"], max_lines=2, overflow=ft.TextOverflow.ELLIPSIS),
            )
            for post in state.posts
        ],
        expand=True,
    )

与 FastAPI 后端集成

Flet 前端与 FastAPI 后端可以在同一进程中运行，也可以分开部署。

同进程运行（开发和小型项目）：

# main.py
import flet as ft
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import threading

# FastAPI 应用
api = FastAPI()

api.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


@api.get("/api/items")
async def get_items():
    return [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]


# Flet 前端
def flet_main(page: ft.Page):
    page.title = "Flet + FastAPI"

    @ft.component
    def App() -> ft.Control:
        return ft.Text("Hello from Flet + FastAPI")

    page.render(App)


def run_api():
    uvicorn.run(api, host="0.0.0.0", port=8000)


if __name__ == "__main__":
    # 在后台线程运行 FastAPI
    api_thread = threading.Thread(target=run_api, daemon=True)
    api_thread.start()

    # 主线程运行 Flet
    ft.app(flet_main)

本地文件读写

import flet as ft
import json
import os


@ft.observable
@dataclass
class FileState:
    content: str = ""
    saved: bool = True


@ft.component
def FileEditor(filename: str) -> ft.Control:
    state, _ = ft.use_state(FileState())

    def load():
        if os.path.exists(filename):
            with open(filename, "r", encoding="utf-8") as f:
                state.content = f.read()

    def save(e):
        with open(filename, "w", encoding="utf-8") as f:
            f.write(state.content)
        state.saved = True

    loaded, set_loaded = ft.use_state(False)
    if not loaded:
        set_loaded(True)
        load()

    return ft.Column([
        ft.Row([
            ft.Text(filename, weight=ft.FontWeight.BOLD),
            ft.Text(
                "已保存" if state.saved else "未保存",
                color=ft.Colors.GREEN if state.saved else ft.Colors.ORANGE,
            ),
            ft.ElevatedButton("保存", on_click=save),
        ], alignment=ft.MainAxisAlignment.SPACE_BETWEEN),
        ft.TextField(
            value=state.content,
            multiline=True,
            min_lines=20,
            expand=True,
            on_change=lambda e: (
                setattr(state, "content", e.control.value),
                setattr(state, "saved", False),
            ),
        ),
    ], expand=True)

加密敏感数据

import flet as ft
from cryptography.fernet import Fernet
import base64
import hashlib


def derive_key(password: str) -> bytes:
    """从用户密码派生加密密钥。"""
    key = hashlib.sha256(password.encode()).digest()
    return base64.urlsafe_b64encode(key)


def encrypt(data: str, password: str) -> str:
    f = Fernet(derive_key(password))
    return f.encrypt(data.encode()).decode()


def decrypt(token: str, password: str) -> str:
    f = Fernet(derive_key(password))
    return f.decrypt(token.encode()).decode()


# 在 client_storage 中存储加密数据
async def save_secret(page: ft.Page, key: str, value: str, password: str):
    encrypted = encrypt(value, password)
    await page.client_storage.set_async(key, encrypted)


async def load_secret(page: ft.Page, key: str, password: str) -> str:
    encrypted = await page.client_storage.get_async(key)
    if encrypted:
        return decrypt(encrypted, password)
    return ""

认证与安全

OAuth2 / 第三方登录

Flet 通过 page.launch_url() 打开浏览器完成 OAuth 授权，再由后端接收回调。

import flet as ft
from dataclasses import dataclass
import asyncio
import httpx


@ft.observable
@dataclass
class AuthState:
    access_token: str = ""
    user_info: dict = None
    is_loading: bool = False
    error: str = ""


@ft.component
def LoginPage(page: ft.Page, auth_state: AuthState) -> ft.Control:

    async def start_oauth():
        auth_state.is_loading = True
        auth_state.error = ""
        # 打开授权页面（后端生成 URL 并返回）
        async with httpx.AsyncClient() as client:
            resp = await client.get("http://localhost:8000/auth/url")
            data = resp.json()
        page.launch_url(data["url"])
        # 等待后端通过 PubSub 发来 token
        auth_state.is_loading = False

    if auth_state.is_loading:
        return ft.Column(
            [ft.ProgressRing(), ft.Text("授权中，请在浏览器中完成...")],
            horizontal_alignment=ft.CrossAxisAlignment.CENTER,
        )

    return ft.Column(
        controls=[
            ft.Text("请登录", size=32, weight=ft.FontWeight.BOLD),
            ft.ElevatedButton(
                "使用 GitHub 登录",
                icon=ft.Icons.CODE,
                on_click=lambda e: asyncio.create_task(start_oauth()),
            ),
            ft.Text(auth_state.error, color=ft.Colors.RED_400) if auth_state.error else ft.Container(),
        ],
        horizontal_alignment=ft.CrossAxisAlignment.CENTER,
        alignment=ft.MainAxisAlignment.CENTER,
        expand=True,
        spacing=16,
    )

Session Storage vs Client Storage

async def save_preference(page: ft.Page, key: str, value):
    await page.client_storage.set_async(key, value)


async def load_preference(page: ft.Page, key: str, default=None):
    val = await page.client_storage.get_async(key)
    return val if val is not None else default


# 示例：保存主题偏好
async def apply_saved_theme(page: ft.Page):
    theme = await load_preference(page, "theme_mode", "system")
    page.theme_mode = {
        "light": ft.ThemeMode.LIGHT,
        "dark": ft.ThemeMode.DARK,
        "system": ft.ThemeMode.SYSTEM,
    }.get(theme, ft.ThemeMode.SYSTEM)
    page.update()

打包与部署

使用 flet build 打包

flet build 依赖 Flutter SDK。首次运行时会自动下载。

# 检查版本
flet --version

# 打包为 Windows 桌面应用
flet build windows

# 打包为 macOS 应用
flet build macos

# 打包为 Android APK
flet build apk

# 打包为 iOS（仅限 macOS）
flet build ipa

# 打包为静态 Web 应用
flet build web

部署到 Web（Docker）

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

COPY pyproject.toml .
COPY src/ ./src/

RUN pip install flet httpx

EXPOSE 8000

CMD ["flet", "run", "--web", "--port", "8000", "src/main.py"]

# docker-compose.yml
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - FLET_SECRET_KEY=your-secret-key
    restart: unless-stopped

Web 应用性能配置

对于大量数据的 Web 应用，需增大 WebSocket 消息限制：

# 启动时设置环境变量
export FLET_WS_MAX_MESSAGE_SIZE=8388608  # 8 MB

flet run --web src/main.py

GitHub Actions CI/CD

# .github/workflows/build.yml
name: Build

on:
  push:
    tags: ["v*"]

jobs:
  build-windows:
    runs-on: windows-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install flet
      - run: flet build windows --output dist/windows
      - uses: actions/upload-artifact@v4
        with:
          name: windows-build
          path: dist/windows/

  build-web:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install flet
      - run: flet build web --output dist/web
      - uses: actions/upload-artifact@v4
        with:
          name: web-build
          path: dist/web/

从零构建复杂应用

本节以「任务管理 + 团队协作」应用为例，展示完整的开发流程。功能包括：

• 任务的增删改查
• 实时多用户协作（PubSub）
• 本地数据持久化
• 深色/浅色主题切换
• 响应式布局（桌面和移动端）

第一步：创建项目结构

mkdir task-manager && cd task-manager

# 创建目录结构
mkdir -p src/{state,components,pages,services}
touch src/main.py
touch src/state/{__init__.py,store.py,models.py}
touch src/components/{__init__.py,navbar.py,sidebar.py}
touch src/pages/{__init__.py,board.py,settings.py}
touch src/services/{__init__.py,storage.py}
touch pyproject.toml

第二步：定义数据模型

# src/state/models.py
from dataclasses import dataclass, field
from typing import Literal
import flet as ft
import time


@ft.observable
@dataclass
class Task:
    id: str = ""
    title: str = ""
    description: str = ""
    status: Literal["todo", "in_progress", "done"] = "todo"
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)


@ft.observable
@dataclass
class Column:
    id: str = ""
    title: str = ""
    tasks: list[Task] = field(default_factory=list)

第三步：创建全局 Store

# src/state/store.py
from dataclasses import dataclass, field
import flet as ft
from .models import Task, Column
import uuid
import time
import json


@ft.observable
@dataclass
class AppStore:
    columns: list[Column] = field(default_factory=lambda: [
        Column(id="todo", title="待办"),
        Column(id="in_progress", title="进行中"),
        Column(id="done", title="已完成"),
    ])
    username: str = "用户"
    theme_mode: str = "system"

    # 广播回调（用于 PubSub 通知其他用户）
    _broadcast: callable = None

    def set_broadcast(self, fn: callable):
        self._broadcast = fn

    def add_task(self, title: str, column_id: str = "todo"):
        task = Task(
            id=str(uuid.uuid4())[:8],
            title=title,
            status=column_id,
            created_at=time.time(),
        )
        col = next((c for c in self.columns if c.id == column_id), None)
        if col:
            col.tasks = col.tasks + [task]
        if self._broadcast:
            self._broadcast({"action": "add_task", "task": self._task_to_dict(task)})

    def move_task(self, task_id: str, from_col: str, to_col: str):
        src = next((c for c in self.columns if c.id == from_col), None)
        dst = next((c for c in self.columns if c.id == to_col), None)
        if not src or not dst:
            return
        task = next((t for t in src.tasks if t.id == task_id), None)
        if not task:
            return
        task.status = to_col
        src.tasks = [t for t in src.tasks if t.id != task_id]
        dst.tasks = dst.tasks + [task]
        if self._broadcast:
            self._broadcast({"action": "move_task", "task_id": task_id, "from": from_col, "to": to_col})

    def delete_task(self, task_id: str):
        for col in self.columns:
            col.tasks = [t for t in col.tasks if t.id != task_id]
        if self._broadcast:
            self._broadcast({"action": "delete_task", "task_id": task_id})

    def apply_remote_event(self, event: dict):
        """处理来自其他用户的实时事件。"""
        action = event.get("action")
        if action == "add_task":
            task_data = event["task"]
            task = Task(**task_data)
            col = next((c for c in self.columns if c.id == task.status), None)
            if col:
                # 避免重复
                if not any(t.id == task.id for t in col.tasks):
                    col.tasks = col.tasks + [task]
        elif action == "move_task":
            self.move_task(event["task_id"], event["from"], event["to"])
        elif action == "delete_task":
            self.delete_task(event["task_id"])

    def save_to_json(self, path: str):
        data = {
            "columns": [
                {"id": c.id, "title": c.title, "tasks": [self._task_to_dict(t) for t in c.tasks]}
                for c in self.columns
            ]
        }
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def load_from_json(self, path: str):
        import os
        if not os.path.exists(path):
            return
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        for col_data in data.get("columns", []):
            col = next((c for c in self.columns if c.id == col_data["id"]), None)
            if col:
                col.tasks = [Task(**t) for t in col_data.get("tasks", [])]

    def _task_to_dict(self, task: Task) -> dict:
        return {
            "id": task.id,
            "title": task.title,
            "description": task.description,
            "status": task.status,
            "created_at": task.created_at,
            "updated_at": task.updated_at,
        }

第四步：看板页面组件

# src/pages/board.py
import flet as ft
from state.store import AppStore


@ft.component
def BoardPage(store: AppStore, page: ft.Page) -> ft.Control:
    # 检测是否为移动端（宽度 < 600）
    is_mobile = page.width < 600 if page.width else False

    columns = [
        KanbanColumn(col=col, store=store)
        for col in store.columns
    ]

    if is_mobile:
        # 移动端：横向可滚动
        return ft.Row(
            controls=columns,
            scroll=ft.ScrollMode.AUTO,
            vertical_alignment=ft.CrossAxisAlignment.START,
        )
    else:
        # 桌面端：三列等宽布局
        return ft.Row(
            controls=columns,
            expand=True,
            vertical_alignment=ft.CrossAxisAlignment.START,
            spacing=16,
        )


@ft.component
def KanbanColumn(col, store: AppStore) -> ft.Control:
    new_title, set_title = ft.use_state("")
    adding, set_adding = ft.use_state(False)

    def confirm_add(e):
        if new_title.strip():
            store.add_task(new_title.strip(), col.id)
        set_title("")
        set_adding(False)

    header = ft.Row(
        controls=[
            ft.Container(
                content=ft.Text(col.title, weight=ft.FontWeight.BOLD, size=14),
                bgcolor=ft.Colors.with_opacity(0.15, ft.Colors.BLUE),
                padding=ft.padding.symmetric(4, 8),
                border_radius=12,
            ),
            ft.Container(
                content=ft.Text(str(len(col.tasks)), size=12, color=ft.Colors.GREY),
                padding=ft.padding.symmetric(2, 6),
            ),
            ft.IconButton(
                icon=ft.Icons.ADD,
                icon_size=16,
                on_click=lambda e: set_adding(True),
                tooltip="添加任务",
            ),
        ],
        alignment=ft.MainAxisAlignment.SPACE_BETWEEN,
    )

    add_form = ft.Column([
        ft.TextField(
            value=new_title,
            hint_text="任务标题",
            autofocus=True,
            on_change=lambda e: set_title(e.control.value),
            on_submit=confirm_add,
        ),
        ft.Row([
            ft.ElevatedButton("添加", on_click=confirm_add),
            ft.TextButton("取消", on_click=lambda e: set_adding(False)),
        ]),
    ]) if adding else ft.Container()

    task_cards = [
        TaskCard(task=task, store=store)
        for task in col.tasks
    ]

    return ft.Container(
        content=ft.Column(
            controls=[header, add_form] + task_cards,
            spacing=8,
            scroll=ft.ScrollMode.AUTO,
        ),
        bgcolor=ft.Colors.with_opacity(0.05, ft.Colors.ON_SURFACE),
        border_radius=12,
        padding=12,
        width=280,
        expand=True,
    )


@ft.component
def TaskCard(task, store: AppStore) -> ft.Control:
    STATUS_LABELS = {"todo": "待办", "in_progress": "进行中", "done": "完成"}
    OTHER_COLS = {
        "todo": ["in_progress", "done"],
        "in_progress": ["todo", "done"],
        "done": ["todo", "in_progress"],
    }

    menu_items = [
        ft.PopupMenuItem(
            text=f"移至：{STATUS_LABELS[col_id]}",
            on_click=lambda e, cid=col_id: store.move_task(task.id, task.status, cid),
        )
        for col_id in OTHER_COLS.get(task.status, [])
    ] + [
        ft.PopupMenuItem(),  # 分隔线
        ft.PopupMenuItem(
            text="删除",
            on_click=lambda e: store.delete_task(task.id),
        ),
    ]

    return ft.Container(
        content=ft.Row([
            ft.Text(task.title, expand=True),
            ft.PopupMenuButton(
                icon=ft.Icons.MORE_VERT,
                icon_size=16,
                items=menu_items,
            ),
        ], alignment=ft.MainAxisAlignment.SPACE_BETWEEN),
        bgcolor=ft.Colors.SURFACE,
        border_radius=8,
        padding=ft.padding.symmetric(8, 12),
        border=ft.border.all(1, ft.Colors.with_opacity(0.1, ft.Colors.ON_SURFACE)),
    )

第五步：导航栏和主布局

# src/components/navbar.py
import flet as ft
from state.store import AppStore


@ft.component
def AppNavBar(store: AppStore, page: ft.Page) -> ft.Control:

    def toggle_theme(e):
        modes = ["system", "light", "dark"]
        idx = modes.index(store.theme_mode)
        store.theme_mode = modes[(idx + 1) % len(modes)]
        page.theme_mode = {
            "light": ft.ThemeMode.LIGHT,
            "dark": ft.ThemeMode.DARK,
            "system": ft.ThemeMode.SYSTEM,
        }[store.theme_mode]
        page.update()

    theme_icon = {
        "light": ft.Icons.LIGHT_MODE,
        "dark": ft.Icons.DARK_MODE,
        "system": ft.Icons.BRIGHTNESS_AUTO,
    }.get(store.theme_mode, ft.Icons.BRIGHTNESS_AUTO)

    return ft.AppBar(
        title=ft.Text("任务看板"),
        center_title=False,
        actions=[
            ft.IconButton(
                icon=theme_icon,
                tooltip="切换主题",
                on_click=toggle_theme,
            ),
            ft.PopupMenuButton(
                icon=ft.Icons.ACCOUNT_CIRCLE,
                items=[
                    ft.PopupMenuItem(text=f"当前用户：{store.username}"),
                    ft.PopupMenuItem(),
                    ft.PopupMenuItem(
                        text="设置",
                        on_click=lambda e: page.go("/settings"),
                    ),
                ],
            ),
        ],
    )

第六步：入口文件整合

# src/main.py
import flet as ft
import re
from state.store import AppStore
from pages.board import BoardPage
from pages.settings import SettingsPage
from components.navbar import AppNavBar

DATA_FILE = "tasks.json"


def main(page: ft.Page):
    page.title = "任务看板"
    page.padding = 0
    page.window.min_width = 360
    page.window.min_height = 600

    store = AppStore()
    store.load_from_json(DATA_FILE)

    # PubSub：向其他用户广播变更
    def broadcast(event: dict):
        page.pubsub.send_all_on_topic("board", event)

    store.set_broadcast(broadcast)

    # 接收其他用户的广播
    def on_remote_event(event: dict):
        store.apply_remote_event(event)

    page.pubsub.subscribe_topic("board", on_remote_event)

    # 关闭时取消订阅并保存
    def on_disconnect(e):
        page.pubsub.unsubscribe_all()
        store.save_to_json(DATA_FILE)

    page.on_disconnect = on_disconnect

    @ft.component
    def Router() -> ft.Control:
        route = page.route or "/"

        if route == "/settings":
            content = SettingsPage(store=store, page=page)
        else:
            content = BoardPage(store=store, page=page)

        return ft.Column(
            controls=[
                AppNavBar(store=store, page=page),
                ft.Container(
                    content=content,
                    expand=True,
                    padding=ft.padding.all(16),
                ),
            ],
            expand=True,
            spacing=0,
        )

    def on_route_change(e: ft.RouteChangeEvent):
        page.render(Router)

    page.on_route_change = on_route_change
    page.render(Router)


ft.app(main)

踩坑与最佳实践

状态更新原则

# 错误：直接修改列表不会触发 observable 通知
state.tasks.append(new_task)   # 不触发重渲染

# 正确：赋值新列表才会触发
state.tasks = state.tasks + [new_task]

避免在渲染函数中产生副作用

# 错误：每次渲染都会重新订阅
@ft.component
def BadComponent(page) -> ft.Control:
    page.pubsub.subscribe(handler)  # 每次渲染都执行！
    return ft.Text("hi")

# 正确：用 use_state 的 "只运行一次" 模式
@ft.component
def GoodComponent(page) -> ft.Control:
    subscribed, set_subscribed = ft.use_state(False)
    if not subscribed:
        set_subscribed(True)
        page.pubsub.subscribe(handler)  # 只运行一次
    return ft.Text("hi")

异步操作错误处理

@ft.component
def SafeAsync() -> ft.Control:
    result, set_result = ft.use_state(None)
    error, set_error = ft.use_state("")
    loading, set_loading = ft.use_state(False)

    async def fetch():
        set_loading(True)
        set_error("")
        try:
            data = await some_api_call()
            set_result(data)
        except Exception as e:
            set_error(str(e))
        finally:
            set_loading(False)

    if loading:
        return ft.ProgressRing()
    if error:
        return ft.Text(error, color=ft.Colors.RED_400)
    if result is None:
        return ft.ElevatedButton("加载", on_click=lambda e: asyncio.create_task(fetch()))
    return ft.Text(str(result))

@ft.component 与 @ft.control 的选择

最佳实践

全局状态用单例 dataclass + page.session：将应用状态定义为一个 @dataclass 实例，存入 page.session["app_state"]，在任意视图中取出修改后调用 page.update()。避免模块级全局变量，因为 Web 多用户场景下会共享状态。

自定义控件（@ft.control）用 is_isolated() 做性能隔离：高频更新的控件（如实时图表、计时器）返回 True 让 Flet 跳过与父控件的 diff，减少不必要的前端渲染。仅对真正独立更新的控件启用，滥用会导致父容器无法触发子控件重渲染。

PubSub 用结构化消息类型而非裸字符串：用 @dataclass 定义消息类型（ChatMessage(user, text)），发布时 page.pubsub.send_all(msg)，订阅回调中用类型检查区分不同消息，避免字符串解析和错误匹配。

与第三方库集成时用 page.run_task 调度协程：在 Flet 事件回调中启动异步任务应用 page.run_task(my_coroutine())，而非直接 asyncio.create_task()。Flet 内部有自己的事件循环，直接调用 asyncio API 可能挂起或与 Flet 调度冲突。

打包前用 flet build 的 --base-url 对齐部署路径：Web 部署到子路径（如 /app/）时，不配置 --base-url=/app/ 会导致静态资源 404。桌面打包时用 --product-name 和 --org 避免与系统已有应用名冲突。

常见陷阱

陷阱：@ft.control 的状态在路由切换后丢失

现象： 导航离开再回来后，自定义控件的内部状态（如展开/折叠）重置为初始值。
原因： Flet 在重新构建视图树时，若控件 key 变化或控件被替换，会创建新实例，旧实例的状态不会保留。
解决： 将需要跨路由保持的状态提升到 page.session 或外部状态对象；给需要保持身份的控件设置稳定的 key。

陷阱：PubSub 订阅在用户断开后仍然存活导致内存泄漏

现象： 多用户 Web 部署下，长时间运行后内存持续增长。
原因： 每个连接订阅了 PubSub 频道，但断开时未取消订阅，订阅回调中的闭包持有 page 对象引用，阻止垃圾回收。
解决： 在 page.on_disconnect 中调用 page.pubsub.unsubscribe_all()；对于定时器等资源也要在 on_disconnect 中取消。

陷阱：flet build web 后某些 Python 包无法使用

现象： 本地运行正常，flet build web 后功能缺失或报 ModuleNotFoundError。
原因： Flet Web 打包使用 Pyodide（WebAssembly 版 Python），不支持所有 CPython 扩展；有 C 扩展的包（如 numpy < 1.26、cryptography）需要 Pyodide 专门构建的版本。
解决： 查阅 Pyodide 兼容包列表；纯 Python 实现的包通常可用；涉及 C 扩展的功能考虑移到后端 API，前端只负责 UI。

参见

Flet入门指南
Flet中级指南
asyncio异步编程完全指南