Python Flet教程（三）：实现数据持久化

noter/
│
├── main.py
├── note.py
├── database.py    # 新增：数据库操作
├── schema.sql     # 新增：数据库结构
├── requirements.txt
└── assets/
    └── icon.png

-- 创建笔记表
CREATE TABLE IF NOT EXISTS notes (
    id TEXT PRIMARY KEY,        -- 使用UUID作为主键
    title TEXT NOT NULL,        -- 笔记标题不能为空
    content TEXT,              -- 笔记内容可以为空
    created_at TIMESTAMP NOT NULL,  -- 创建时间
    updated_at TIMESTAMP NOT NULL   -- 更新时间
);

-- 创建索引以优化按更新时间排序的查询
CREATE INDEX IF NOT EXISTS idx_notes_updated 
ON notes(updated_at DESC);

import sqlite3
from datetime import datetime
from typing import List, Optional
from note import Note
import os

class Database:
    def __init__(self, db_path: str):
        """
        初始化数据库管理类
        Args:
            db_path: 数据库文件路径
        """
        self.db_path = db_path
        self.init_db()

    def init_db(self):
        """
        初始化数据库：
        1. 创建数据库目录（如果不存在）
        2. 创建数据库表和索引
        3. 处理可能的数据库错误
        """
        try:
            # 确保数据库目录存在
            os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
            
            # 创建数据库连接并执行初始化脚本
            with sqlite3.connect(self.db_path) as conn:
                with open('schema.sql', 'r', encoding='utf-8') as f:
                    conn.executescript(f.read())
        except sqlite3.Error as e:
            print(f"数据库初始化错误: {e}")
            raise

    def get_connection(self):
        """
        获取数据库连接
        Returns:
            sqlite3.Connection: 数据库连接对象
        """
        try:
            return sqlite3.connect(self.db_path)
        except sqlite3.Error as e:
            print(f"数据库连接错误: {e}")
            raise

    def save_note(self, note: Note) -> None:
        """
        保存或更新笔记
        Args:
            note: 要保存的笔记对象
        """
        sql = '''
        INSERT OR REPLACE INTO notes (id, title, content, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?)
        '''
        try:
            with self.get_connection() as conn:
                conn.execute(sql, (
                    note.id,
                    note.title,
                    note.content,
                    note.created_at.isoformat(),  # 转换为ISO格式字符串
                    note.updated_at.isoformat()
                ))
        except sqlite3.Error as e:
            print(f"保存笔记错误: {e}")
            raise

    def delete_note(self, note_id: str) -> None:
        """
        删除笔记
        Args:
            note_id: 要删除的笔记ID
        """
        sql = 'DELETE FROM notes WHERE id = ?'
        try:
            with self.get_connection() as conn:
                conn.execute(sql, (note_id,))
        except sqlite3.Error as e:
            print(f"删除笔记错误: {e}")
            raise

    def get_all_notes(self) -> List[Note]:
        """
        获取所有笔记，按更新时间降序排序
        Returns:
            List[Note]: 笔记对象列表
        """
        sql = 'SELECT * FROM notes ORDER BY updated_at DESC'
        try:
            with self.get_connection() as conn:
                cursor = conn.execute(sql)
                return [self._row_to_note(row) for row in cursor.fetchall()]
        except sqlite3.Error as e:
            print(f"获取笔记列表错误: {e}")
            raise

    def search_notes(self, query: str) -> List[Note]:
        """
        搜索笔记
        Args:
            query: 搜索关键词
        Returns:
            List[Note]: 匹配的笔记列表
        """
        sql = '''
        SELECT * FROM notes 
        WHERE title LIKE ? OR content LIKE ?
        ORDER BY updated_at DESC
        '''
        try:
            with self.get_connection() as conn:
                cursor = conn.execute(sql, (f'%{query}%', f'%{query}%'))
                return [self._row_to_note(row) for row in cursor.fetchall()]
        except sqlite3.Error as e:
            print(f"搜索笔记错误: {e}")
            raise

    @staticmethod
    def _row_to_note(row) -> Note:
        """
        将数据库行转换为Note对象
        Args:
            row: 数据库查询结果行
        Returns:
            Note: 转换后的笔记对象
        """
        try:
            return Note(
                id=row[0],
                title=row[1],
                content=row[2],
                created_at=datetime.fromisoformat(row[3]),
                updated_at=datetime.fromisoformat(row[4])
            )
        except (IndexError, ValueError) as e:
            print(f"数据转换错误: {e}")
            raise

from dataclasses import dataclass
from datetime import datetime
import uuid

@dataclass
class Note:
    # ... (之前的代码保持不变)

    def to_dict(self) -> dict:
        """
        将Note对象转换为字典，用于序列化
        Returns:
            dict: 包含笔记数据的字典
        """
        return {
            'id': self.id,
            'title': self.title,
            'content': self.content,
            'created_at': self.created_at.isoformat(),
            'updated_at': self.updated_at.isoformat()
        }

    @classmethod
    def from_dict(cls, data: dict) -> "Note":
        """
        从字典创建Note对象，用于反序列化
        Args:
            data: 包含笔记数据的字典
        Returns:
            Note: 创建的笔记对象
        """
        try:
            return cls(
                id=data['id'],
                title=data['title'],
                content=data['content'],
                created_at=datetime.fromisoformat(data['created_at']),
                updated_at=datetime.fromisoformat(data['updated_at'])
            )
        except (KeyError, ValueError) as e:
            print(f"从字典创建Note对象错误: {e}")
            raise

import flet as ft
from note import Note
from database import Database
import os

class NoterApp:
    def __init__(self, page: ft.Page):
        self.page = page
        self.setup_page()
        
        # 初始化数据库
        db_path = os.path.join(os.path.dirname(__file__), 'data', 'notes.db')
        try:
            self.db = Database(db_path)
            # 从数据库加载笔记
            self.notes: list[Note] = self.db.get_all_notes()
        except Exception as e:
            print(f"数据库初始化错误: {e}")
            # 显示错误提示
            self.page.show_snack_bar(
                ft.SnackBar(content=ft.Text("数据库初始化失败，请检查数据库文件"))
            )
            self.notes = []
        
        self.current_note: Note = None
        self.setup_controls()
        self.render()

    # ... (保持其他方法不变)

    def create_note(self, e=None):
        """创建新笔记并保存到数据库"""
        try:
            note = Note.create("新建笔记", "")
            self.notes.append(note)
            self.db.save_note(note)  # 保存到数据库
            self.select_note(note)
            self.update_notes_list()
            self.title_field.focus()
            self.page.update()
        except Exception as e:
            print(f"创建笔记错误: {e}")
            self.page.show_snack_bar(
                ft.SnackBar(content=ft.Text("创建笔记失败"))
            )

    def delete_note(self, note: Note):
        if note in self.notes:
            self.notes.remove(note)
            self.db.delete_note(note.id)  # 从数据库删除
            if self.current_note == note:
                self.current_note = None
                self.title_field.value = ""
                self.content_field.value = ""
            self.update_notes_list()
            self.page.update()

    def on_title_change(self, e):
        if self.current_note:
            self.current_note.update(title=e.control.value)
            self.db.save_note(self.current_note)  # 保存到数据库
            self.update_notes_list()
            self.page.update()

    def on_content_change(self, e):
        """处理笔记内容变化并保存"""
        if self.current_note:
            try:
                self.current_note.update(content=e.control.value)
                self.db.save_note(self.current_note)
                self.update_notes_list()
                self.page.update()
            except Exception as e:
                print(f"保存笔记错误: {e}")
                self.page.show_snack_bar(
                    ft.SnackBar(content=ft.Text("保存笔记失败"))
                )

    def search_notes(self, e):
        query = e.control.value.lower()
        if query:
            # 使用数据库搜索
            self.notes = self.db.search_notes(query)
        else:
            # 查询为空时显示所有笔记
            self.notes = self.db.get_all_notes()
        self.update_notes_list()
        self.page.update()

    def render(self):
        # 主布局
        self.page.add(
            ft.Row(
                controls=[
                    self.sidebar,
                    ft.Container(
                        content=self.notes_list,
                        width=300,
                        bgcolor=ft.colors.with_opacity(0.5, ft.colors.BLUE_GREY_50),
                    ),
                    self.editor,
                ],
                expand=True,
            )
        )
        # 确保所有控件都已添加到页面后，更新笔记列表
        self.page.update()
        self.update_notes_list()