⏳ Async Operations

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]

1. Luôn hiển thị loading - Cho user biết đang xử lý
2. Handle errors - Wrap trong try/except
3. Cho phép cancel - Với long tasks
4. Tránh block UI - Dùng async cho I/O operations
5. 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.