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

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

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

相关推荐

大文件传不动?WinRAR/7-Zip 入门到高手,这 5 个技巧让你效率翻倍

“这200张照片怎么传给女儿?微信发不了,邮箱附件又超限……”62岁的张阿姨对着电脑犯愁时,儿子只用了3分钟就把照片压缩成一个文件,还教她:“以后用压缩软件,比打包行李还方便!”职场人更懂这...

电脑解压缩软件推荐——7-Zip:免费、高效、简洁的文件管理神器

在日常工作中,我们经常需要处理压缩文件。无论是下载软件包、接收文件,还是存储大量数据,压缩和解压缩文件都成为了我们日常操作的一部分。而说到压缩解压软件,7-Zip绝对是一个不可忽视的名字。今天,我就来...

设置了加密密码zip文件要如何打开?这几个方法可以试试~

Zip是一种常见的压缩格式文件,文件还可以设置密码保护。那设置了密码的Zip文件要如何打开呢?不清楚的小伙伴一起来看看吧。当我们知道密码想要打开带密码的Zip文件,我们需要用到适用于Zip格式的解压缩...

大文件想要传输成功,怎么把ZIP文件分卷压缩

不知道各位小伙伴有没有这样的烦恼,发送很大很大的压缩包会受到限制,为此,想要在压缩过程中将文件拆分为几个压缩包并且同时为所有压缩包设置加密应该如何设置?方法一:使用7-Zip免费且强大的文件管理工具7...

高效处理 RAR 分卷压缩包:合并解压操作全攻略

在文件传输和存储过程中,当遇到大文件时,我们常常会使用分卷压缩的方式将其拆分成多个较小的压缩包,方便存储和传输。RAR作为一种常见的压缩格式,分卷压缩包的使用频率也很高。但很多人在拿到RAR分卷...

2个方法教你如何删除ZIP压缩包密码

zip压缩包设置了加密密码,每次解压文件都需要输入密码才能够顺利解压出文件,当压缩包文件不再需要加密的时候,大家肯定想删除压缩包密码,或是忘记了压缩包密码,想要通过删除操作将压缩包密码删除,就能够顺利...

速转!漏洞预警丨压缩软件Winrar目录穿越漏洞

WinRAR是一款功能强大的压缩包管理器,它是档案工具RAR在Windows环境下的图形界面。该软件可用于备份数据,缩减电子邮件附件的大小,解压缩从Internet上下载的RAR、ZIP及其它类...

文件解压方法和工具分享_文件解压工具下载

压缩文件减少文件大小,降低文件失效的概率,总得来说好处很多。所以很多文件我们下载下来都是压缩软件,很多小伙伴不知道怎么解压,或者不知道什么工具更好,所以今天做了文件解压方法和工具的分享给大家。一、解压...

[python]《Python编程快速上手:让繁琐工作自动化》学习笔记3

1.组织文件笔记(第9章)(代码下载)1.1文件与文件路径通过importshutil调用shutil模块操作目录,shutil模块能够在Python程序中实现文件复制、移动、改名和删除;同时...

Python内置tarfile模块:读写 tar 归档文件详解

一、学习目标1.1学习目标掌握Python内置模块tarfile的核心功能,包括:理解tar归档文件的原理与常见压缩格式(gzip/bz2/lzma)掌握tar文件的读写操作(创建、解压、查看、过滤...

使用python展开tar包_python拓展

类Unix的系统,打包文件经常使用的就是tar包,结合zip工具,可以方便的打包并解压。在python的标准库里面有tarfile库,可以方便实现生成了展开tar包。使用这个库最大的好处,可能就在于不...

银狐钓鱼再升级:白文件脚本化实现GO语言后门持久驻留

近期,火绒威胁情报中心监测到一批相对更为活跃的“银狐”系列变种木马。火绒安全工程师第一时间获取样本并进行分析。分析发现,该样本通过阿里云存储桶下发恶意文件,采用AppDomainManager进行白利...

ZIP文件怎么打开?2个简单方法教你轻松搞定!

在日常工作和生活中,我们经常会遇到各种压缩文件,其中最常见的格式之一就是ZIP。ZIP文件通过压缩数据来减少文件大小,方便我们进行存储和传输。然而,对于初学者来说,如何打开ZIP文件可能会成为一个小小...

Ubuntu—解压多个zip压缩文件.zip .z01 .z02

方法将所有zip文件放在同一目录中:zip_file.z01,zip_file.z02,zip_file.z03,...,zip_file.zip。在Zip3.0版本及以上,使用下列命令:将所有zi...

如何使用7-Zip对文件进行加密压缩

7-Zip是一款开源的文件归档工具,支持多种压缩格式,并提供了对压缩文件进行加密的功能。使用7-Zip可以轻松创建和解压.7z、.zip等格式的压缩文件,并且可以通过设置密码来保护压缩包中的...

取消回复欢迎 发表评论: