主機端 Flask REST APIs 設計

後端程式我們使用 flask + flask_restful 來完成我們的 REST API, 為什麼不使用 nodejs 來完成呢？nodejs 語言對我來說是最不適合來寫程式，主要邏輯中一大堆不必要的迴圈一大堆的非線性處理，用你最上手的程式語言就好，二來 python 還有很多優秀的 libraries，可以處理很多事情。

測試時使用 sqlite 資料庫，也可以替換成其他 SQL 資料庫。

結論

pandas 幾乎在其他語言中沒有替代品，已經用 python 開發出各種工具，開發速度也比之前使用 ruby 還要方便直覺，財報分析、股票績效計算、Schweb 轉 MyStocksApp、繁轉轉換等都是透過 python 完成，目前陸續整合成 API ，陸續會開放出來。常用的 library 如下表：

pandas: 資料統計與分析，程式語言界的 excel
numpy: 與 pandas 搭配, The fundamental package for scientific computing with Python.
flask: 前後端網站開發
metaflow: Netflix 開源的軟體，他的主要目的是資料整理的時候，可以分拆成很多個步驟，例如在 MyFlowSpec 步驟a 發現有 bugs 或是需要整理出不同格式的數據， MyFlowSpec 修改後可以產生新的數據，在 MyFlowSpec 之前與之後的程式都可以不用變動，加快專案或是數據探索流程。

數據整理或分析統計原始流程

修改後的流程：左半邊是原始程式產生數據Ａ，右半邊是修改後程式產生數據Ｂ修改後流程

OpenBB: OpenBBTerminal 開源版的彭博終端機 bloomberg terminal

flask 介紹

flask 網站開發 framework
flask_restful 快速開發 API
flask_sqlalchemy 簡單快速上手使用 sqlalchemy library 支援各種主流的 SQL 資料庫

流程

前端在上一篇已經完成，這邊使用 python 來完成後端系統，依照功能與用途分別有五個步驟，分別放在不同的目錄之下，可以看下圖目錄架構。

# 後端目錄架構
src          #       
  project    #    
    app      # 主程式
    handlers # api 
    models   # model 設定
    settings # 系統設定
  tests      # 測試

flask_sqlalchemy 定義 User 的資料類別

使用 sql 的資料庫最麻煩的一點就是只要表格有變動，資料庫也要變動，不然完全不能使用。

在 models 目錄下新增 user.py 檔案，這裡很簡單就只是定義 User SQL table 欄位。

z5525/models/user.py

# z5525/models/user.py
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=False, nullable=False)
    email = db.Column(db.String(100), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)
    is_admin = db.Column(db.Boolean, default=False)  # 管理員權限

    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

    def to_dict(self):
        d = {
            "id": self.id,
            "username": self.username,
            "email": self.email,
            "is_admin": self.is_admin,
        }
        return d

一樣是在 models 目錄新增 __init__.py 將 db 與 User 匯入到這個檔案之中，之後要引用 from z5525.models import db, User 就可以使用

z5525/models/__init__.py

# z5525/models/__init__.py
from z5525.models.user import db, User

利用 flask_restful 來完成 API

1.寫 UserResource 用來讀取、新增、更新、刪除資料庫，現在只先完成 get user 和 users，最後面再補上新增、更增、刪除等。

z5525/handlers/users.py

# z5525/handlers/users.py
from flask import Flask, request, jsonify
from flask_restful import Api, Resource, abort
from werkzeug.security import generate_password_hash, check_password_hash
from z5525.models.user import db, User
import sqlalchemy.exc as sa_exc

# 
class UserResource(Resource):

    # READ
    def get(self, user_id=None):
        if user_id:
            try:
                user = db.session.execute(db.select(User).filter_by(id=user_id)).scalar_one()
                return {
                    "status": "ok",
                    "total": 1,
                    "count": 1,
                    'user': user.to_dict()
                }, 200
            except sa_exc.NoResultFound:
                return {
                    "error": "Not Found", 
                    "message": 'User not found'
                }, 404
        else:
            total = db.session.query(User).count()  # 總數量
            offset = request.args.get('offset', default=0, type=int)
            limit = request.args.get('limit', default=10, type=int)
            users = db.session.execute(db.select(User).offset(offset).limit(limit)).scalars().all()
            return {
                "status": "ok",
                "total": total,
                "count": len(users),
                "offset": offset,
                "limit": limit,
                'items': [user.to_dict() for user in users]
            }, 200

2.定義 API

z5525/handlers/__init__.py

# z5525/handlers/__init__.py
from flask_restful import Api, Resource
from milch.handlers.users import UserResource

APP_PREFIX = "/api"

api = Api()
api.add_resource(UserResource, f'{APP_PREFIX}/users', f'{APP_PREFIX}/users/<int:user_id>')

主程式：最後一個步驟

命名為 z5525/app/main.py

z5525/app/main.py

# z5525/app/main.py
from flask import Flask, request, jsonify
from z5525.models import db, User
from z5525.handlers import api

def create_app():
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    app.json.compact = False
    db.init_app(app) 
    api.init_app(app)
    return app   

if __name__ == '__main__':
    app = create_app()
    with app.app_context():
        db.create_all()    
    app.run(port=5525)

最後補上 __init__.py 裡面只加入一行

z5525/app/__init__.py

# z5525/app/__init__.py
from .main import create_app

測試 python -m z5525.app.main

 * Serving Flask app 'main'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5525

看到以上的訊息表示已經起動，可以開始測試。打開瀏覽器 http://127.0.0.1:5525/api/users 或用 curl http://127.0.0.1:5525/api/users 指令去測試 User API 。

curl http://127.0.0.1:5525/api/users
{"status": "ok", "total": 0, "count": 0, "offset": 0, "limit": 10, "items": []}

pytest 寫測試

在目錄新增 user 測試，檔名命名為 test_user.py

1.測試 client

以下兩種都可以正常使用的 client，使用 create_app 的好處是，測試與正式上線的設定是一樣的，當有任何設定變更只需要在 settings 中變更，例如、目前範例都是寫死的，可以將這些設定統一到 settings 。

# 使用 create_app
@pytest.fixture
def client():    
    """ 建立 Flask 測試客戶端 """
    app = create_app() # 在 z5525/app/main.py  
    with app.app_context():
        db.create_all()
        yield app.test_client()

# 使用全新的測試環境
@pytest.fixture
def client():
    """ 建立 Flask 測試客戶端 """
    app = Flask(__name__)
    app.config['TESTING'] = True
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'  # 使用記憶體內的 SQLite
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    db.init_app(app)
    api = Api(app)
    api.add_resource(UserResource, '/users', '/users/<int:user_id>')

    with app.app_context():
        db.create_all()
        yield app.test_client()

2.測試數據

和在參數中加入它後，每次的測試都是乾淨的，只要有加入會影響到測試的資料，反過來沒有加入的，測試數據庫是不會有的。

@pytest.fixture
def sample_user():
    """ 插入一筆測試 """
    user = User(
        username="Test user",
        email="[email protected]",
        password="pass",
    )
    db.session.add(user)
    db.session.commit()
    return user

3.測試 get users

參數中加入了和，因為已經加入 sample_user ，在這次的測試中數據庫裡面已經有一筆數據，這筆數據就是 sample_user 所產生的。

def test_get_users(client, sample_user):
    """ 測試取得 users API，並檢查 JSON 格式 """
    response = client.get(f'{APP_PREFIX}/users')
    assert response.status_code == 200
    assert "total" in response.json
    assert "count" in response.json
    assert "offset" in response.json
    assert "limit" in response.json
    assert "items" in response.json
    assert len(response.json["items"]) == 1

完整 test_user.py

將 pytest 測試的三種合起來就是一個測試檔案，將它寫在同一個檔案裡面，就會如下程式碼

tests/test_user.py

# tests/test_user.py
import pytest
import os
from flask import Flask
from z5525.app.main import create_app
from z5525.handlers import api, UserResource
from z5525.models import db, User

APP_PREFIX = "/api"

@pytest.fixture
def client():
    """ 建立 Flask 測試客戶端 """
    app = create_app() # 在 z5525/app/main.py
    with app.app_context():
        db.create_all()
        yield app.test_client()

@pytest.fixture
def sample_user():
    """ 插入一筆測試產品 """
    user = User(
        username="Test user",
        email="[email protected]",
        password="pass",
    )
    db.session.add(user)
    db.session.commit()
    return user

def test_get_users(client, sample_user):
    """ 測試取得 users API，並檢查 JSON 格式 """
    response = client.get(f'{APP_PREFIX}/users')
    assert response.status_code == 200
    assert "total" in response.json
    assert "count" in response.json
    assert "offset" in response.json
    assert "limit" in response.json
    assert "items" in response.json
    assert len(response.json["items"]) == 1

def test_get_users_limit(client, sample_user):
    """ 測試取得 users API，並檢查 JSON 格式 """
    response = client.get(f'{APP_PREFIX}/users?offset=0&limit=1')
    assert response.status_code == 200
    assert "total" in response.json
    assert "count" in response.json
    assert "offset" in response.json
    assert "limit" in response.json
    assert "items" in response.json
    assert len(response.json["items"]) == 1

def test_get_user(client, sample_user):
    """ 測試取得 user API """
    response = client.get(f'{APP_PREFIX}/users/{sample_user.id}')
    assert response.status_code == 200
    assert "total" in response.json
    assert "count" in response.json
    assert "offset" in response.json
    assert "limit" in response.json
    assert "items" in response.json
    assert response.json["total"] == 1
    assert response.json["count"] == 1
    assert response.json["user"]["username"] == "Test user"

設定

一定要補上設定檔，不然執行會找不到你寫的 module

pyproject.toml

# pyproject.toml
[tool.pytest.ini_options]
pythonpath = ["./"]

開始測試

pytest tests/test_user.py 或是 pytest

User model 補上新增、刪除與更新

z5525/models/user.py

# z5525/models/user.py
class UserResource(Resource):
    # CREATE
    def post(self):
        try:
            data = request.get_json(silent=True)
        except Exception as e:
            return {
                "error": "Bad Request", 
                "message": "Invalid JSON format in request body", 
            }, 400

        username = data.get('username')
        email = data.get('email')
        password = data.get('password')
        if not password or not email:
            return {
                "error": "Bad Request", 
                "message": 'password and email are required'}, 400
        
        try:
            user = db.session.execute(db.select(User).filter_by(email=email)).scalar_one()
            return {
                "error": "Bad Request", 
                "message": 'User already exists'}, 400
        except sa_exc.NoResultFound:
            """ Not Found: Create it """

        hashed_password = generate_password_hash(password)
        new_user = User(username=username, email=email, password=hashed_password)
        db.session.add(new_user)
        db.session.commit()
        return {
            "status": "ok", 
            'message': 'User added successfully'}, 201

    # UPDATE
    def put(self, user_id):
        data = request.get_json(silent=True)
        username = data.get('username')
        email = data.get('email')
        if not username or not email:
            return {
                "error": "Bad Request", 
                "message": 'Username and email are required'}, 400

        try:
            user = db.session.execute(db.select(User).filter_by(id=user_id)).scalar_one()
            user.username = username
            user.email = email
            db.session.commit()
            return {
                "status": "ok", 
                "message": 'User updated successfully'}, 200
        except sa_exc.NoResultFound:
            return {
                "error": "Not Found", 
                "message": 'User not found'}, 404

    # DELETE
    def delete(self, user_id):
        try:
            user = db.session.execute(db.select(User).filter_by(id=user_id)).scalar_one()
            db.session.delete(user)
            db.session.commit()
            return {
                "status": "ok", 
                "message": 'User deleted successfully'}, 200
        except sa_exc.NoResultFound:
            return {
                "error": "Not Found", 
                "message": 'User not found'}, 404

test User 補上新增、刪除與更新等測試

tests/test_user.py

# tests/test_user.py
def test_create_user(client):
    """ 測試新增 user API """
    response = client.post(f'{APP_PREFIX}/users', json={
        "username": "New Phone",
        "email": "[email protected]",
        "password": "pass",
    })
    assert response.status_code == 201
    assert response.json["status"] == "ok"
    users = User.query.all()
    assert len(users) == 1


def test_update_user(client, sample_user):
    """ 測試更新 user API """
    response = client.put(f'{APP_PREFIX}/users/{sample_user.id}', json={"username": 'hello', "email": "[email protected]"})
    assert response.status_code == 200
    assert response.json["status"] == "ok"

def test_delete_user(client, sample_user):
    """ 測試 DELETE user API """
    response = client.delete(f'{APP_PREFIX}/users/{sample_user.id}')
    assert response.status_code == 200
    assert response.json["status"] == "ok"

def test_deleted_get_user(client, sample_user):
    """ 測試 DELETE user 後再查詢，應該回傳 404 """
    client.delete(f'{APP_PREFIX}/users/{sample_user.id}')
    response = client.get(f'/users/{sample_user.id}')
    assert response.status_code == 404

參考網站 Reference

pandas
numpy
Flask
flask_restful
flask_sqlalchemy
OpenBB
Metaflow
Best practices for RESTful web API design

版本備註

python 3.11
flask 3.0.3
flask_restful 0.3.10
flask_sqlalchemy 3.1.1
pytest 8.4.1