百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术资源 > 正文

Python 进阶-day24: API 开发(python的api)

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

学习目标

  1. 理解 RESTful API 的核心概念和设计原则。
  2. 使用 Flask 创建模块化的 RESTful API,包含优雅的数据库访问代码。
  3. 为博客应用实现 API 接口,支持 CRUD 操作(创建、读取、更新、删除)。
  4. 学习 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:设置开发环境

  1. 安装依赖: 确保安装了 Flask 和 Flask-SQLAlchemy:bash
  2. pip install flask flask-sqlalchemy
  3. 项目结构: 创建以下目录结构,确保代码模块化:



  1. 生成 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

  1. 运行 Flask 应用:bash服务器运行在 http://127.0.0.1:5000。
python app.py
  1. 使用工具测试: 使用 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
  1. 获取所有文章
curl http://127.0.0.1:5000/api/posts
  1. 获取单篇文章:bash
curl http://127.0.0.1:5000/api/posts/1
  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
  1. 删除文章:bash
curl -X DELETE http://127.0.0.1:5000/api/posts/1
  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 功能

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

  1. 添加用户认证

使用 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():
    # 现有代码
  1. 支持分页

在 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
  1. 添加搜索功能

在 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
  1. 添加统一错误处理

使用 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 上下文的依赖。
  • 一致性:统一的错误处理和数据访问接口,减少重复代码。
  • 扩展性:支持分页、搜索等功能只需扩展对应模块。

相关推荐

pip的使用及配置_pip怎么配置

要使用python必须要学会使用pip,pip的全称:packageinstallerforpython,也就是Python包管理工具,主要是对python的第三方库进行安装、更新、卸载等操作,...

Anaconda下安装pytorch_anaconda下安装tensorflow

之前的文章介绍了tensorflow-gpu的安装方法,也介绍了许多基本的工具与使用方法,具体可以看Ubuntu快速安装tensorflow2.4的gpu版本。pytorch也是一个十分流行的机器学...

Centos 7 64位安装 python3的教程

wgethttps://www.python.org/ftp/python/3.10.13/Python-3.10.13.tgz#下载指定版本软件安装包tar-xzfPython-3.10.1...

如何安装 pip 管理工具_pip安装详细步骤

如何安装pip管理工具方法一:yum方式安装Centos安装python3和python3-devel开发包>#yuminstallgcclibffi-develpy...

Python入门——从开发环境搭建到hello world

一、Python解释器安装1、在windows下步骤1、下载安装包https://www.python.org/downloads/打开后选择【Downloads】->【Windows】小编是一...

生产环境中使用的十大 Python 设计模式

在软件开发的浩瀚世界中,设计模式如同指引方向的灯塔,为我们构建稳定、高效且易于维护的系统提供了经过验证的解决方案。对于Python开发者而言,理解和掌握这些模式,更是提升代码质量、加速开发进程的关...

如何创建和管理Python虚拟环境_python怎么创建虚拟环境

在Python开发中,虚拟环境是隔离项目依赖的关键工具。下面介绍创建和管理Python虚拟环境的主流方法。一、内置工具:venv(Python3.3+推荐)venv是Python标准...

初学者入门Python的第一步——环境搭建

Python如今成为零基础编程爱好者的首选学习语言,这和Python语言自身的强大功能和简单易学是分不开的。今天千锋武汉Python培训小编将带领Python零基础的初学者完成入门的第一步——环境搭建...

全网最简我的世界Minecraft搭建Python编程环境

这篇文章将给大家介绍一种在我的世界minecraft里搭建Python编程开发环境的操作方法。目前看起来应该是全网最简单的方法。搭建完成后,马上就可以利用python代码在我的世界自动创建很多有意思的...

Python开发中的虚拟环境管理_python3虚拟环境

Python开发中,虚拟环境管理帮助隔离项目依赖,避免不同项目之间的依赖冲突。虚拟环境的作用隔离依赖:不同项目可能需要不同版本的库,虚拟环境可以为每个项目创建独立的环境。避免全局污染:全局安装的库可...

Python内置zipfile模块:操作 ZIP 归档文件详解

一、知识导图二、知识讲解(一)zipfile模块概述zipfile模块是Python内置的用于操作ZIP归档文件的模块。它提供了创建、读取、写入、添加及列出ZIP文件的功能。(二)ZipFile类1....

Python内置模块pydoc :文档生成器和在线帮助系统详解

一、引言在Python开发中,良好的文档是提高代码可读性和可维护性的关键。pydoc是Python自带的一个强大的文档生成器和在线帮助系统,它可以根据Python模块自动生成文档,并支持多种输出格式...

Python sys模块使用教程_python system模块

1.知识导图2.sys模块概述2.1模块定义与作用sys模块是Python标准库中的一个内置模块,提供了与Python解释器及其环境交互的接口。它包含了许多与系统相关的变量和函数,可以用来控制P...

Python Logging 模块完全解读_python logging详解

私信我,回复:学习,获取免费学习资源包。Python中的logging模块可以让你跟踪代码运行时的事件,当程序崩溃时可以查看日志并且发现是什么引发了错误。Log信息有内置的层级——调试(deb...

软件测试|Python logging模块怎么使用,你会了吗?

Pythonlogging模块使用在开发和维护Python应用程序时,日志记录是一项非常重要的任务。Python提供了内置的logging模块,它可以帮助我们方便地记录应用程序的运行时信息、错误和调...

取消回复欢迎 发表评论: