Python 进阶-day24: API 开发（pythonapi接口开发）

liftword1周前 (05-02)技术文章7

学习目标

理解 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 模式和服务层，数据库访问代码具有以下优势：

分离关注点：路由（视图）、服务（业务逻辑）、Repository（数据访问）职责清晰。
模块化：数据库操作封装在 Repository 层，易于维护和扩展。
可测试性：服务层和 Repository 层可以独立测试，减少对 Flask 上下文的依赖。
一致性：统一的错误处理和数据访问接口，减少重复代码。
扩展性：支持分页、搜索等功能只需扩展对应模块。

Python小技巧:如何批量更新已安装的库?

众所周知，升级某个库（假设为 xxx），可以用pip install --upgrade xxx 命令，或者简写成pip install -U xxx 。如果有多个库，可以依次写在 xxx 后面，以空...

主编推荐 | Gurobi 并行计算的设置和操作(附代码)

『运筹OR帷幄』原创作者：运筹OR帷幄编者按实际应用问题往往具有较高的计算复杂度，而优化算法难以在实际中落地的主要瓶颈就在于无法满足实际问题对计算时间的苛刻要求。然而近年来随着计算力的蓬勃发展，并行计...

升级到PyTorch 2.0的技巧总结（pytorch的最新版本）

来源：DeepHub IMBA本文约6400字，建议阅读12分钟在本文将演示 PyTorch 2.0新功能的使用，以及介绍在使用它时可能遇到的一些问题。PyTorch 2.0 发布也有一段时间了，大家...

提升 Python 代码水平的 5 个方法

这是原始函数：from datetime import datetime def is_legal_age(dob_str): """Checks if a pers...

Python pip升级及升级失败解决方案

本教程用于Python pip升级及失败解决方案首先查看脚本pip show pip我已经升级到了最新的版本安装其他模块过程中出现下面提示,便说明你需要升级pipYou are using pip v...

零基础AI开发系列教程:Dify升级指南

Dify近期发布很是频繁，基本两三天一个版本。值得肯定的是优化和改进了很多问题，但是官方的升级文档有点分散，也有点乱。我这里整理了一个升级文档供大家参考，如果还没有升级到新版本的小伙伴，可以按照我的文...

流照教程网

Python 进阶-day24: API 开发（pythonapi接口开发）

相关文章

Python小技巧:如何批量更新已安装的库?

主编推荐 | Gurobi 并行计算的设置和操作(附代码)

升级到PyTorch 2.0的技巧总结（pytorch的最新版本）

提升 Python 代码水平的 5 个方法

Python pip升级及升级失败解决方案

零基础AI开发系列教程:Dify升级指南

蜀ICP备2024111239号-1