Python 进阶-day24: API 开发(python的api)
off999 2025-05-08 04:41 15 浏览 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 模式和服务层,数据库访问代码具有以下优势:
- 分离关注点:路由(视图)、服务(业务逻辑)、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模块,它可以帮助我们方便地记录应用程序的运行时信息、错误和调...
你 发表评论:
欢迎- 一周热门
- 最近发表
- 标签列表
-
- python计时 (73)
- python安装路径 (56)
- python类型转换 (93)
- python进度条 (67)
- python吧 (67)
- python的for循环 (65)
- python格式化字符串 (61)
- python静态方法 (57)
- python列表切片 (59)
- python面向对象编程 (60)
- python 代码加密 (65)
- python串口编程 (77)
- python封装 (57)
- python写入txt (66)
- python读取文件夹下所有文件 (59)
- python操作mysql数据库 (66)
- python获取列表的长度 (64)
- python接口 (63)
- python调用函数 (57)
- python多态 (60)
- python匿名函数 (59)
- python打印九九乘法表 (65)
- python赋值 (62)
- python异常 (69)
- python元祖 (57)