当前位置：网站首页 > 技术资源 > 正文

Python 进阶-day24: API 开发（python的api）

off999 2025-05-08 04:41 37 浏览 0 评论

学习目标

理解 RESTful API 的核心概念和设计原则。
使用 Flask 创建模块化的 RESTful API，包含优雅的数据库访问代码。
为博客应用实现 API 接口，支持 CRUD 操作（创建、读取、更新、删除）。
学习 Repository 模式和服务层，优化数据库操作的组织方式。

步骤 1：理解 RESTful API

REST（Representational State Transfer）是一种基于 HTTP 的架构风格，用于设计网络 API。核心特点：

资源（Resources）：通过 URL 表示资源（如 /api/posts 表示博客文章）。
HTTP 方法：使用 GET、POST、PUT、DELETE 等方法操作资源。
无状态：每个请求包含所有必要信息，服务器不存储客户端状态。
JSON 格式：通常以 JSON 格式交换数据。

示例：

GET /api/posts：获取所有文章。
POST /api/posts：创建新文章。
PUT /api/posts/1：更新 ID 为 1 的文章。
DELETE /api/posts/1：删除 ID 为 1 的文章。

步骤 2：设置开发环境

安装依赖：确保安装了 Flask 和 Flask-SQLAlchemy：bash
pip install flask flask-sqlalchemy
项目结构：创建以下目录结构，确保代码模块化：

生成 requirements.txt：bash

pip freeze > requirements.txt

步骤 3：实现博客的 RESTful API

我们将创建一个博客 API，支持文章的 CRUD 操作，使用 Repository 模式和服务层分离数据库操作和业务逻辑。

3.1 配置（config.py）

python

# config.py
class Config:
    SQLALCHEMY_DATABASE_URI = 'sqlite:///blog.db'  # 使用 SQLite 数据库
    SQLALCHEMY_TRACK_MODIFICATIONS = False

3.2 定义模型（models.py）

定义博客文章的数据库模型，并提供序列化方法。

python

# models.py
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)

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

3.3 创建 Repository 层（post_repository.py）

Repository 层封装数据库操作，提供统一的接口。

python

# repositories/post_repository.py
from models import Post, db
from sqlalchemy.exc import IntegrityError

class PostRepository:
    def get_all(self):
        """获取所有文章"""
        return Post.query.all()

    def get_by_id(self, post_id):
        """根据 ID 获取文章"""
        return Post.query.get(post_id)

    def create(self, title, content):
        """创建新文章"""
        post = Post(title=title, content=content)
        db.session.add(post)
        try:
            db.session.commit()
            return post
        except IntegrityError:
            db.session.rollback()
            raise ValueError("Failed to create post due to database error")

    def update(self, post_id, title, content):
        """更新文章"""
        post = self.get_by_id(post_id)
        if not post:
            return None
        post.title = title
        post.content = content
        try:
            db.session.commit()
            return post
        except IntegrityError:
            db.session.rollback()
            raise ValueError("Failed to update post due to database error")

    def delete(self, post_id):
        """删除文章"""
        post = self.get_by_id(post_id)
        if not post:
            return False
        db.session.delete(post)
        try:
            db.session.commit()
            return True
        except IntegrityError:
            db.session.rollback()
            raise ValueError("Failed to delete post due to database error")

3.4 创建服务层（post_service.py）

服务层处理业务逻辑，调用 Repository 层并进行输入验证。

python

# services/post_service.py
from repositories.post_repository import PostRepository

class PostService:
    def __init__(self):
        self.repository = PostRepository()

    def get_all_posts(self):
        """获取所有文章"""
        posts = self.repository.get_all()
        return [post.to_dict() for post in posts]

    def get_post(self, post_id):
        """获取单篇文章"""
        post = self.repository.get_by_id(post_id)
        if not post:
            raise ValueError("Post not found")
        return post.to_dict()

    def create_post(self, title, content):
        """创建新文章"""
        if not title or not content:
            raise ValueError("Title and content are required")
        post = self.repository.create(title, content)
        return post.to_dict()

    def update_post(self, post_id, title, content):
        """更新文章"""
        if not title or not content:
            raise ValueError("Title and content are required")
        post = self.repository.update(post_id, title, content)
        if not post:
            raise ValueError("Post not found")
        return post.to_dict()

    def delete_post(self, post_id):
        """删除文章"""
        success = self.repository.delete(post_id)
        if not success:
            raise ValueError("Post not found")
        return {"message": "Post deleted"}

3.5 创建 Flask 应用（app.py）

路由函数调用服务层，保持简洁。

python

# app.py
from flask import Flask, request, jsonify
from config import Config
from models import db
from services.post_service import PostService

# 初始化 Flask 应用
app = Flask(__name__)
app.config.from_object(Config)
db.init_app(app)

# 创建数据库表
with app.app_context():
    db.create_all()

# 初始化服务层
post_service = PostService()

# 获取所有文章
@app.route('/api/posts', methods=['GET'])
def get_posts():
    try:
        posts = post_service.get_all_posts()
        return jsonify(posts)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

# 获取单篇文章
@app.route('/api/posts/<int:id>', methods=['GET'])
def get_post(id):
    try:
        post = post_service.get_post(id)
        return jsonify(post)
    except ValueError as e:
        return jsonify({'error': str(e)}), 404
    except Exception as e:
        return jsonify({'error': str(e)}), 500

# 创建新文章
@app.route('/api/posts', methods=['POST'])
def create_post():
    try:
        data = request.get_json()
        post = post_service.create_post(data.get('title'), data.get('content'))
        return jsonify(post), 201
    except ValueError as e:
        return jsonify({'error': str(e)}), 400
    except Exception as e:
        return jsonify({'error': str(e)}), 500

# 更新文章
@app.route('/api/posts/<int:id>', methods=['PUT'])
def update_post(id):
    try:
        data = request.get_json()
        post = post_service.update_post(id, data.get('title'), data.get('content'))
        return jsonify(post)
    except ValueError as e:
        return jsonify({'error': str(e)}), 400
    except Exception as e:
        return jsonify({'error': str(e)}), 500

# 删除文章
@app.route('/api/posts/<int:id>', methods=['DELETE'])
def delete_post(id):
    try:
        result = post_service.delete_post(id)
        return jsonify(result)
    except ValueError as e:
        return jsonify({'error': str(e)}), 404
    except Exception as e:
        return jsonify({'error': str(e)}), 500

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

步骤 4：测试 API

运行 Flask 应用：bash服务器运行在 http://127.0.0.1:5000。

python app.py

使用工具测试：使用 Postman、cURL 或 Python 的 requests 库。例如：

curl -X POST -H "Content-Type: application/json" -d '{"title":"My Post","content":"This is my first post"}' http://127.0.0.1:5000/api/posts

获取所有文章：

curl http://127.0.0.1:5000/api/posts

获取单篇文章：bash

curl http://127.0.0.1:5000/api/posts/1

更新文章：bash

curl -X PUT -H "Content-Type: application/json" -d '{"title":"Updated Post","content":"Updated content"}' http://127.0.0.1:5000/api/posts/1

删除文章：bash

curl -X DELETE http://127.0.0.1:5000/api/posts/1

使用 Python 测试（可选）：创建 test_api.py：python运行：bash

import requests

base_url = 'http://127.0.0.1:5000/api/posts'

# 创建文章
response = requests.post(base_url, json={'title': 'Test Post', 'content': 'This is a test'})
print(response.json())

# 获取所有文章
response = requests.get(base_url)
print(response.json())

步骤 5：练习任务 - 扩展 API 功能

基于上述代码，完成以下练习以加深理解：

添加用户认证：

使用 flask-jwt-extended 或 flask-httpauth 实现 API 认证，限制只有登录用户可以创建/更新/删除文章。

安装：pip install flask-jwt-extended

示例：python

from flask_jwt_extended import JWTManager, jwt_required

app.config['JWT_SECRET_KEY'] = 'your-secret-key'
jwt = JWTManager(app)

@app.route('/api/posts', methods=['POST'])
@jwt_required()
def create_post():
    # 现有代码

支持分页：

在 PostRepository 中添加分页方法：python

def get_paginated(self, page, per_page):
    return Post.query.paginate(page=page, per_page=per_page)

PostService 中调用：python

def get_all_posts(self, page=1, per_page=10):
    pagination = self.repository.get_paginated(page, per_page)
    return {
        'posts': [post.to_dict() for post in pagination.items],
        'total': pagination.total,
        'pages': pagination.pages
    }

更新路由：python

@app.route('/api/posts', methods=['GET'])
def get_posts():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    try:
        result = post_service.get_all_posts(page, per_page)
        return jsonify(result)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

添加搜索功能：

在 PostRepository 中添加搜索方法：python

def search(self, keyword):
    return Post.query.filter(Post.title.contains(keyword) | Post.content.contains(keyword)).all()

在 PostService 中调用：python

def search_posts(self, keyword):
    posts = self.repository.search(keyword)
    return [post.to_dict() for post in posts]

更新路由：python

@app.route('/api/posts', methods=['GET'])
def get_posts():
    keyword = request.args.get('q')
    try:
        if keyword:
            posts = post_service.search_posts(keyword)
        else:
            posts = post_service.get_all_posts()
        return jsonify(posts)
    except Exception as e:
        return jsonify({'error': str(e)}), 500

添加统一错误处理：

使用 Flask 的错误处理器：python

@app.errorhandler(ValueError)
def handle_value_error(error):
    return jsonify({'error': str(error)}), 400

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

为什么代码更优雅？

通过引入 Repository 模式和服务层，数据库访问代码具有以下优势：