当前位置：首页 > article >正文

使用Flask构建RESTful API

article 2025/1/10 23:36:37

💓 博客主页：瑕疵的CSDN主页

📝 Gitee主页：瑕疵的gitee主页

⏩ 文章专栏：《热点资讯》

使用Flask构建RESTful API

- - Flask简介
  - 环境搭建
  - - 安装Flask
  - 项目结构
  - 创建应用
  - 路由定义
  - 请求处理
  - - 获取查询参数
    - 获取请求体
  - 响应格式化
  - - JSON响应
  - 错误处理
  - 数据库集成
  - - 安装SQLAlchemy
    - 配置数据库
    - 定义模型
    - 初始化数据库
  - 测试
  - 总结

Flask是一个轻量级的Web框架，适用于快速开发小型到中型的Web应用。本文将详细介绍如何使用Flask构建RESTful API，包括环境搭建、项目结构、路由定义、请求处理、响应格式化、错误处理、数据库集成、测试等内容。

Flask简介

Flask是一个用Python编写的轻量级Web应用框架。它没有固定的数据库抽象层、表单验证工具等，因此非常灵活，适合快速开发。

环境搭建

在开始之前，确保你的环境中已安装Python和pip。

安装Flask

pip install Flask

项目结构

一个典型的Flask项目结构如下：

my_flask_app/
├── app.py
├── config.py
├── models.py
├── routes.py
└── requirements.txt

创建应用

在`app.py`中创建Flask应用实例。

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/hello', methods=['GET'])
def hello():
    return jsonify({'message': 'Hello, World!'}), 200

if __name__ == '__main__':
    app.run(debug=True)

路由定义

在`routes.py`中定义路由。

from flask import Blueprint, jsonify, request

api_bp = Blueprint('api', __name__)

@api_bp.route('/users', methods=['GET'])
def get_users():
    users = [
        {'id': 1, 'name': 'Alice'},
        {'id': 2, 'name': 'Bob'}
    ]
    return jsonify(users), 200

@api_bp.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = {'id': user_id, 'name': 'Unknown'}
    return jsonify(user), 200

@api_bp.route('/users', methods=['POST'])
def create_user():
    data = request.json
    user = {'id': 3, 'name': data['name']}
    return jsonify(user), 201

# 导入蓝图
from app import app
app.register_blueprint(api_bp, url_prefix='/api')

请求处理

在Flask中，可以使用`request`对象来处理HTTP请求。

获取查询参数

@app.route('/search', methods=['GET'])
def search():
    query = request.args.get('q', '')
    results = []
    return jsonify(results), 200

获取请求体

@app.route('/submit', methods=['POST'])
def submit():
    data = request.json
    result = process_data(data)
    return jsonify(result), 200

响应格式化

Flask提供了多种方式来格式化响应。

JSON响应

@app.route('/json', methods=['GET'])
def json_response():
    data = {'key': 'value'}
    return jsonify(data), 200

错误处理

可以使用`@app.errorhandler`装饰器来定义全局错误处理程序。

@app.errorhandler(404)
def not_found(error):
    return jsonify({'error': 'Not found'}), 404

@app.errorhandler(500)
def internal_error(error):
    return jsonify({'error': 'Internal server error'}), 500

数据库集成

可以使用SQLAlchemy来集成数据库。

安装SQLAlchemy

pip install SQLAlchemy

配置数据库

在 config.py中配置数据库。

import os

basedir = os.path.abspath(os.path.dirname(__file__))

SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False

定义模型

在 models.py中定义数据库模型。

from flask_sqlalchemy import SQLAlchemy
from config import SQLALCHEMY_DATABASE_URI

app.config['SQLALCHEMY_DATABASE_URI'] = SQLALCHEMY_DATABASE_URI
db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), index=True, unique=True)

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name
        }

初始化数据库

在 app.py中初始化数据库。

from models import db

db.init_app(app)
with app.app_context():
    db.create_all()

测试

可以使用Flask的测试客户端来编写单元测试。

import unittest
from app import app

累加器 = 0

class TestApp(unittest.TestCase):
    def setUp(self):
        self.app = app.test_client()
        self.ctx = app.app_context()
        self.ctx.push()

    def tearDown(self):
        self.ctx.pop()

    def test_hello(self):
        response = self.app.get('/hello')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json, {'message': 'Hello, World!'})

    def test_get_users(self):
        response = self.app.get('/api/users')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(len(response.json), 2)

if __name__ == '__main__':
    unittest.main()