财务经理教程 - 第3部分

在本教程的这一部分中，我们将扩展我们的财务管理应用程序，使用FastAPI和Uvicorn创建一个REST API。这将使我们能够执行服务器端操作，并与在第2部分中使用[SQLAlchemy]创建的SQLite数据库进行交互。

FastAPI 是一个现代的、快速的（高性能的）Web框架，用于使用Python构建API。 它建立在[ASGI（异步服务器网关接口）]之上，这使得它能够高效地处理并发请求。

Uvicorn 是一个极快的 ASGI 服务器实现，使用 uvloop 和 httptools。它为运行 FastAPI 应用程序提供了高性能的服务器。

要下载本教程的完整源代码，请访问 Finance Manager Example - Part 3。

先决条件

在我们开始之前，请确保您的Python环境中已安装FastAPI和Uvicorn。

你可以使用pip安装它们：

pip install fastapi uvicorn

项目结构

本教程这一部分的整体项目结构与之前的部分有很大不同。我们将使用PySide6和QML的前端部分移动到一个名为Frontend的单独目录中，而将使用FastAPI创建REST API到SQLite数据库的后端部分移动到一个名为Backend的目录中。

├── Backend
│   ├── database.py
│   ├── main.py
│   └── rest_api.py
├── Frontend
│   ├── Finance
│   │   ├── AddDialog.qml
│   │   ├── FinanceDelegate.qml
│   │   ├── FinancePieChart.qml
│   │   ├── FinanceView.qml
│   │   ├── Main.qml
│   │   └── qmldir
│   ├── financemodel.py
│   └── main.py

让我们开始吧！

首先，让我们创建Backend目录，并将previous part中的database.py移动到Backend目录。

后端设置

设置 FastAPI

在Backend目录中创建一个新的Python文件rest_api.py，并添加以下代码：

rest_api.py
rest_api.py
 1# Copyright (C) 2024 The Qt Company Ltd.
 2# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
 3
 4import logging
 5from fastapi import FastAPI, Depends, HTTPException
 6from pydantic import BaseModel
 7from typing import Dict, Any
 8from sqlalchemy import orm
 9from database import Session, Finance
10
11app = FastAPI()
12
13
14class FinanceCreate(BaseModel):
15    item_name: str
16    category: str
17    cost: float
18    date: str
19
20
21class FinanceRead(FinanceCreate):
22    class Config:
23        from_attributes = True
24
25
26def get_db():
27    db = Session()
28    try:
29        yield db
30    finally:
31        db.close()
32
33
34@app.post("/finances/", response_model=FinanceRead)
35def create_finance(finance: FinanceCreate, db: orm.Session = Depends(get_db)):
36    print(f"Adding finance item: {finance}")
37    db_finance = Finance(**finance.model_dump())
38    db.add(db_finance)
39    db.commit()
40    db.refresh(db_finance)
41    return db_finance
42
43
44@app.get("/finances/", response_model=Dict[str, Any])
45def read_finances(skip: int = 0, limit: int = 10, db: orm.Session = Depends(get_db)):
46    try:
47        total = db.query(Finance).count()
48        finances = db.query(Finance).offset(skip).limit(limit).all()
49        response = {
50            "total": total,
51            # Convert the list of Finance objects to a list of FinanceRead objects
52            "items": [FinanceRead.from_orm(finance) for finance in finances]
53        }
54        logging.info(f"Response: {response}")
55        return response
56    except Exception as e:
57        logging.error(f"Error occurred: {e}")
58        raise HTTPException(status_code=500, detail="Internal Server Error")

在rest_api.py中，我们设置了一个FastAPI应用程序来处理财务数据的创建和读取操作。该文件包括：

  1. FastAPI 应用: 初始化一个 FastAPI 实例。

  2. Pydantic 模型: 定义了用于输入验证和创建新记录的 FinanceCreate 以及用于输出格式化的 FinanceRead。FinanceRead 包括对 ORM 模型的额外配置。

  3. 数据库依赖: 提供了一个 get_db 函数来管理数据库会话。

  4. 创建端点: 一个POST端点 /finances/ 用于向数据库中添加新的财务条目。

  5. 读取端点: 一个GET端点 /finances/ 用于从数据库中检索财务条目。

此设置允许应用程序与数据库交互，通过RESTful API端点实现财务记录的创建和检索。

为后端创建一个主要的Python文件

在Backend目录中创建一个新的Python文件main.py，并添加以下代码：

main.py
main.py
 1# Copyright (C) 2024 The Qt Company Ltd.
 2# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
 3
 4import uvicorn
 5from database import initialize_database
 6
 7
 8def main():
 9    # Initialize the database
10    initialize_database()
11    # Start the FastAPI endpoint
12    uvicorn.run("rest_api:app", host="127.0.0.1", port=8000, reload=True)
13
14
15if __name__ == "__main__":
16    main()

在main.py中，除了初始化数据库外，我们还创建了一个Uvicorn服务器实例来运行FastAPI应用程序。服务器运行在127.0.0.1的8000端口，并在检测到代码更改时自动重新加载。

前端设置

然后我们继续到应用程序的前端部分，并将其连接到FastAPI后端。 我们将创建一个新目录Frontend。将文件夹Finance和文件financemodel.py 以及main.py从之前的部分移动到Frontend目录。

大部分代码与之前的部分保持不变，只有一些小的改动以连接到REST API。

更新FinanceModel类

以下更改（已高亮显示）是对financemodel.py文件进行的，这些更改从Python中移除了数据库交互代码，并添加了REST API交互代码。

financemodel.py
financemodel.py
  1# Copyright (C) 2024 The Qt Company Ltd.
  2# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
  3
  4import requests
  5from datetime import datetime
  6from dataclasses import dataclass
  7from enum import IntEnum
  8from collections import defaultdict
  9
 10from PySide6.QtCore import (QAbstractListModel, QEnum, Qt, QModelIndex, Slot,
 11                            QByteArray)
 12from PySide6.QtQml import QmlElement
 13
 14QML_IMPORT_NAME = "Finance"
 15QML_IMPORT_MAJOR_VERSION = 1
 16
 17
 18@QmlElement
 19class FinanceModel(QAbstractListModel):
 20
 21    @QEnum
 22    class FinanceRole(IntEnum):
 23        ItemNameRole = Qt.DisplayRole
 24        CategoryRole = Qt.UserRole
 25        CostRole = Qt.UserRole + 1
 26        DateRole = Qt.UserRole + 2
 27        MonthRole = Qt.UserRole + 3
 28
 29    @dataclass
 30    class Finance:
 31        item_name: str
 32        category: str
 33        cost: float
 34        date: str
 35
 36        @property
 37        def month(self):
 38            return datetime.strptime(self.date, "%d-%m-%Y").strftime("%B %Y")
 39
 40    def __init__(self, parent=None) -> None:
 41        super().__init__(parent)
 42        self.m_finances = []
 43        self.fetchAllData()
 44
 45    def fetchAllData(self):
 46        response = requests.get("http://127.0.0.1:8000/finances/")
 47        try:
 48            data = response.json()
 49        except requests.exceptions.JSONDecodeError:
 50            print("Failed to decode JSON response")
 51            return
 52        self.beginInsertRows(QModelIndex(), 0, len(data["items"]) - 1)
 53        self.m_finances.extend([self.Finance(**item) for item in data["items"]])
 54        self.endInsertRows()
 55
 56    def rowCount(self, parent=QModelIndex()):
 57        return len(self.m_finances)
 58
 59    def data(self, index: QModelIndex, role: int):
 60        if not index.isValid() or index.row() >= self.rowCount():
 61            return None
 62        row = index.row()
 63        if row < self.rowCount():
 64            finance = self.m_finances[row]
 65            if role == FinanceModel.FinanceRole.ItemNameRole:
 66                return finance.item_name
 67            if role == FinanceModel.FinanceRole.CategoryRole:
 68                return finance.category
 69            if role == FinanceModel.FinanceRole.CostRole:
 70                return finance.cost
 71            if role == FinanceModel.FinanceRole.DateRole:
 72                return finance.date
 73            if role == FinanceModel.FinanceRole.MonthRole:
 74                return finance.month
 75        return None
 76
 77    def roleNames(self):
 78        roles = super().roleNames()
 79        roles[FinanceModel.FinanceRole.ItemNameRole] = QByteArray(b"item_name")
 80        roles[FinanceModel.FinanceRole.CategoryRole] = QByteArray(b"category")
 81        roles[FinanceModel.FinanceRole.CostRole] = QByteArray(b"cost")
 82        roles[FinanceModel.FinanceRole.DateRole] = QByteArray(b"date")
 83        roles[FinanceModel.FinanceRole.MonthRole] = QByteArray(b"month")
 84        return roles
 85
 86    @Slot(int, result='QVariantMap')
 87    def get(self, row: int):
 88        finance = self.m_finances[row]
 89        return {"item_name": finance.item_name, "category": finance.category,
 90                "cost": finance.cost, "date": finance.date}
 91
 92    @Slot(str, str, float, str)
 93    def append(self, item_name: str, category: str, cost: float, date: str):
 94        finance = {"item_name": item_name, "category": category, "cost": cost, "date": date}
 95        response = requests.post("http://127.0.0.1:8000/finances/", json=finance)
 96        if response.status_code == 200:
 97            finance = response.json()
 98            self.beginInsertRows(QModelIndex(), 0, 0)
 99            self.m_finances.insert(0, self.Finance(**finance))
100            self.endInsertRows()
101        else:
102            print("Failed to add finance item")
103
104    @Slot(result=dict)
105    def getCategoryData(self):
106        category_data = defaultdict(float)
107        for finance in self.m_finances:
108            category_data[finance.category] += finance.cost
109        return dict(category_data)

为FinanceModel类重写了两个方法 - fetchMore和canFetchMore。这些方法用于在模型滚动到底部时从REST API获取更多数据。数据每次以10个条目为单位进行获取。此外，append方法被更新为向REST API发送POST请求以添加新的财务条目。

更新前端的主Python文件

最后，我们更新了Frontend目录中的main.py文件，以移除数据库初始化代码。

main.py
main.py
 1# Copyright (C) 2024 The Qt Company Ltd.
 2# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
 3
 4import sys
 5from pathlib import Path
 6
 7from PySide6.QtWidgets import QApplication
 8from PySide6.QtQml import QQmlApplicationEngine
 9
10from financemodel import FinanceModel  # noqa: F401
11
12if __name__ == '__main__':
13    app = QApplication(sys.argv)
14    QApplication.setOrganizationName("QtProject")
15    QApplication.setApplicationName("Finance Manager")
16    engine = QQmlApplicationEngine()
17
18    engine.addImportPath(Path(__file__).parent)
19    engine.loadFromModule("Finance", "Main")
20
21    if not engine.rootObjects():
22        sys.exit(-1)
23
24    exit_code = app.exec()
25    del engine
26    sys.exit(exit_code)

其余的代码和QML文件与教程的前几部分保持不变。

运行应用程序

要运行应用程序，首先通过在Backend目录中使用Python执行main.py文件来启动后端FastAPI服务器：

python Backend/main.py

这将在http://127.0.0.1:8000上启动运行FastAPI应用程序的Uvicorn服务器。

启动后端服务器后，通过在Frontend目录中使用Python执行main.py文件来运行前端应用程序：

python Frontend/main.py

这将启动连接到FastAPI后端的PySide6应用程序，以显示财务数据。

部署

部署应用程序

要部署应用程序，请按照教程第一部分中的相同步骤操作。

总结

在本教程结束时，您已经扩展了财务经理应用程序，以包含使用FastAPI和Uvicorn的REST API。后端应用程序使用SQLAlchemy与SQLite数据库交互，并提供创建和检索财务记录的端点。前端应用程序通过REST API连接到后端，以显示财务数据。

如果你想进一步扩展应用程序，你可以添加更多功能，如更新和删除财务记录、添加用户认证等。