Async Operations trong Flet

Học cách xử lý bất đồng bộ trong ứng dụng Flet. ⏱️ 15 phút

🎯 Mục tiêu

Nắm vững:

Async event handlers
Long-running tasks
Progress indicators
Background tasks

⚡ Tại sao cần Async?

Khi thực hiện các tác vụ mất thời gian (API calls, file I/O, tính toán nặng), bạn cần async để:

Không block UI
Hiển thị loading indicators
Cho phép cancel operations

🔄 Async với Declarative UI

Kết hợp @ft.observable với async handlers:


import flet as ft
import asyncio
 
@ft.observable
class LoadingState:
    is_loading: bool = False
    status: str = "Ready"
    
    def start_loading(self):
        self.is_loading = True
        self.status = "Loading..."
    
    def finish_loading(self, message: str = "Done! ✅"):
        self.is_loading = False
        self.status = message
 
state = LoadingState()
 
@ft.component
def AsyncDemo():
    async def fetch_data(_):
        state.start_loading()
        await asyncio.sleep(2)  # Simulate API call
        state.finish_loading("Data loaded! ✅")
    
    return ft.Column([
        ft.ElevatedButton("Fetch Data", on_click=fetch_data),
        ft.ProgressRing(visible=state.is_loading),
        ft.Text(state.status),
    ], spacing=15)
 
async def main(page: ft.Page):
    page.title = "Async Demo"
    page.padding = 30
    page.add(AsyncDemo())
 
ft.run(main)

📊 Progress Indicator


import flet as ft
import asyncio
 
@ft.observable
class ProgressState:
    progress: float = 0
    status: str = "0%"
    is_running: bool = False
    
    def update(self, value: float):
        self.progress = value
        self.status = f"{int(value * 100)}%"
    
    def complete(self):
        self.is_running = False
        self.status = "Done! ✅"
    
    def start(self):
        self.is_running = True
        self.progress = 0
 
state = ProgressState()
 
@ft.component
def ProgressDemo():
    async def start_process(_):
        state.start()
        for i in range(101):
            state.update(i / 100)
            await asyncio.sleep(0.03)
        state.complete()
    
    return ft.Column([
        ft.ProgressBar(value=state.progress, width=300),
        ft.Text(state.status, size=18),
        ft.ElevatedButton(
            "Start", 
            on_click=start_process,
            disabled=state.is_running,
        ),
    ], spacing=15, horizontal_alignment=ft.CrossAxisAlignment.CENTER)
 
async def main(page: ft.Page):
    page.title = "Progress Demo"
    page.vertical_alignment = ft.MainAxisAlignment.CENTER
    page.horizontal_alignment = ft.CrossAxisAlignment.CENTER
    page.add(ProgressDemo())
 
ft.run(main)

🌐 HTTP Requests với httpx


import flet as ft
import httpx
import asyncio
 
@ft.observable
class ApiState:
    is_loading: bool = False
    result: str = ""
    error: str = ""
    
    def start_fetch(self):
        self.is_loading = True
        self.result = ""
        self.error = ""
    
    def set_result(self, data: str):
        self.is_loading = False
        self.result = data
    
    def set_error(self, error: str):
        self.is_loading = False
        self.error = error
 
state = ApiState()
 
@ft.component
def ApiDemo():
    async def fetch_user(_):
        state.start_fetch()
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    "https://jsonplaceholder.typicode.com/users/1"
                )
                data = response.json()
                state.set_result(f"Name: {data['name']}\nEmail: {data['email']}")
        except Exception as e:
            state.set_error(f"Error: {e}")
    
    return ft.Column([
        ft.ElevatedButton("Fetch User", on_click=fetch_user),
        ft.ProgressRing(visible=state.is_loading),
        ft.Text(state.result, selectable=True) if state.result else None,
        ft.Text(state.error, color=ft.Colors.RED) if state.error else None,
    ], spacing=15)
 
async def main(page: ft.Page):
    page.title = "API Demo"
    page.padding = 30
    page.add(ApiDemo())
 
ft.run(main)

⏸️ Cancelable Tasks


import flet as ft
import asyncio
 
@ft.observable
class TaskState:
    status: str = "Idle"
    is_running: bool = False
    task = None
    
    def set_status(self, status: str):
        self.status = status
    
    def start(self):
        self.is_running = True
    
    def stop(self):
        self.is_running = False
 
state = TaskState()
 
@ft.component
def CancelableTaskDemo():
    async def long_task():
        state.start()
        try:
            for i in range(10):
                state.set_status(f"Processing... {i+1}/10")
                await asyncio.sleep(1)
            state.set_status("Completed! ✅")
        except asyncio.CancelledError:
            state.set_status("Cancelled ❌")
        finally:
            state.stop()
    
    async def start_task(_):
        state.task = asyncio.create_task(long_task())
    
    async def cancel_task(_):
        if state.task:
            state.task.cancel()
    
    return ft.Column([
        ft.Row([
            ft.ElevatedButton("Start", on_click=start_task, disabled=state.is_running),
            ft.ElevatedButton("Cancel", on_click=cancel_task, disabled=not state.is_running),
        ]),
        ft.ProgressRing(visible=state.is_running),
        ft.Text(state.status, size=18),
    ], spacing=15)
 
async def main(page: ft.Page):
    page.title = "Cancelable Task"
    page.padding = 30
    page.add(CancelableTaskDemo())
 
ft.run(main)

🔄 Multiple Concurrent Tasks


import flet as ft
import asyncio
 
@ft.observable
class MultiTaskState:
    results: list = []
    is_loading: bool = False
    
    def clear(self):
        self.results = []
        self.is_loading = True
    
    def add_result(self, result: str):
        self.results = self.results + [result]
    
    def complete(self):
        self.is_loading = False
 
state = MultiTaskState()
 
@ft.component
def ResultsList():
    if not state.results:
        return ft.Text("No results yet")
    return ft.Column([
        ft.Text(r, color=ft.Colors.GREEN) for r in state.results
    ])
 
@ft.component
def MultiTaskDemo():
    async def fetch_one(name: str, delay: float):
        await asyncio.sleep(delay)
        state.add_result(f"✅ {name} loaded")
    
    async def fetch_all(_):
        state.clear()
        await asyncio.gather(
            fetch_one("Users", 1.0),
            fetch_one("Products", 1.5),
            fetch_one("Orders", 0.8),
        )
        state.add_result("🎉 All done!")
        state.complete()
    
    return ft.Column([
        ft.ElevatedButton("Fetch All", on_click=fetch_all, disabled=state.is_loading),
        ft.ProgressRing(visible=state.is_loading),
        ResultsList(),
    ], spacing=15)
 
async def main(page: ft.Page):
    page.title = "Multi Task Demo"
    page.padding = 30
    page.add(MultiTaskDemo())
 
ft.run(main)

⏰ Periodic Updates (Clock)


import flet as ft
import asyncio
from datetime import datetime
 
@ft.observable
class ClockState:
    time: str = ""
    is_running: bool = True
    
    def update_time(self):
        self.time = datetime.now().strftime("%H:%M:%S")
 
state = ClockState()
 
@ft.component
def Clock():
    return ft.Column([
        ft.Text("Current Time:", size=20),
        ft.Text(state.time, size=48, weight=ft.FontWeight.BOLD),
    ], horizontal_alignment=ft.CrossAxisAlignment.CENTER)
 
async def main(page: ft.Page):
    page.title = "Clock"
    page.vertical_alignment = ft.MainAxisAlignment.CENTER
    page.horizontal_alignment = ft.CrossAxisAlignment.CENTER
    
    page.add(Clock())
    
    # Background task for clock updates
    async def update_clock():
        while state.is_running:
            state.update_time()
            await asyncio.sleep(1)
    
    asyncio.create_task(update_clock())
 
ft.run(main)

💡 Best Practices

Tip

Luôn hiển thị loading - Cho user biết đang xử lý
Handle errors - Wrap trong try/except
Cho phép cancel - Với long tasks
Tránh block UI - Dùng async cho I/O operations
Throttle updates - Không update quá nhanh

📋 Async Patterns

Pattern	Use Case
`async def handler`	Event handlers với I/O
`asyncio.sleep()`	Delays, animations
`asyncio.gather()`	Multiple concurrent tasks
`asyncio.create_task()`	Background tasks
`task.cancel()`	Cancelable operations

⏭️ Tiếp theo

Tuyệt! Tiếp tục với bài cuối: Build & Deploy để đóng gói và phân phối ứng dụng.