Python实现RESTful API的基础教程

一、RESTful API 简介

REST（Representational State Transfer）即表述性状态转移，是一种设计网络应用程序的架构风格。RESTful API 是遵循 REST 原则设计的应用程序编程接口。

REST 的核心概念围绕资源展开，资源可以是任何有意义的事物，比如一篇文章、一个用户等。每个资源都通过唯一的 URL（Uniform Resource Locator）进行标识。客户端通过 HTTP 协议的不同方法（如 GET、POST、PUT、DELETE 等）与服务器进行交互，来操作这些资源。

例如，对于一个用户资源，可能有如下的 URL 设计：

• GET /users：获取所有用户列表
• GET /users/{id}：获取特定 ID 的用户
• POST /users：创建一个新用户
• PUT /users/{id}：更新特定 ID 的用户信息
• DELETE /users/{id}：删除特定 ID 的用户

这种设计风格使得 API 具有良好的可读性、可扩展性和可维护性，方便不同平台的客户端进行交互。

二、Python 开发 RESTful API 的常用框架

1. Flask Flask 是一个轻量级的 Python Web 框架，它提供了简单的路由系统和请求处理机制，非常适合快速搭建 RESTful API。它的核心依赖是 Werkzeug（WSGI 工具集）和 Jinja2（模板引擎），不过在开发 RESTful API 时，模板引擎并非必需。

2. Django Django 是一个功能强大的全栈 Web 框架，自带了 ORM（对象关系映射）、管理界面、表单处理等众多功能。虽然它相对 Flask 更重量级，但对于大型项目和需要复杂业务逻辑的 RESTful API 开发非常有优势。它遵循“约定大于配置”的原则，能让开发者快速上手。

3. FastAPI FastAPI 是一个基于 Python 的快速 Web 框架，用于构建 API。它使用 Python 的类型提示来提高代码的可读性和可维护性，并且性能出色，基于 ASGI（Asynchronous Server Gateway Interface），支持异步编程，能处理高并发场景。

三、使用 Flask 构建 RESTful API

1. 安装 Flask 首先，确保你已经安装了 Python。然后可以使用 pip 安装 Flask：

pip install flask

2. 创建基本的 Flask 应用 创建一个新的 Python 文件，例如 app.py，编写如下代码：

from flask import Flask

app = Flask(__name__)


@app.route('/')
def hello_world():
    return 'Hello, World!'


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

在上述代码中，我们首先导入 Flask 类，然后创建了一个 Flask 应用实例 app。通过 @app.route 装饰器定义了一个根路由，当访问根路径时，返回“Hello, World!”。app.run(debug=True) 启动了应用，并开启调试模式，方便在开发过程中查看错误信息。

3. 实现简单的 RESTful API 端点 假设我们要创建一个管理书籍的 RESTful API，首先定义书籍的数据结构，这里简单使用列表来模拟数据库：

from flask import Flask, jsonify, request

app = Flask(__name__)

books = [
    {
        'id': 1,
        'title': 'Python Crash Course',
        'author': 'Eric Matthes'
    },
    {
        'id': 2,
        'title': 'Clean Code',
        'author': 'Robert C. Martin'
    }
]


@app.route('/books', methods=['GET'])
def get_books():
    return jsonify(books)


@app.route('/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
    book = next((book for book in books if book['id'] == book_id), None)
    if book is None:
        return jsonify({'message': 'Book not found'}), 404
    return jsonify(book)


@app.route('/books', methods=['POST'])
def create_book():
    data = request.get_json()
    new_book = {
        'id': len(books) + 1,
        'title': data.get('title'),
        'author': data.get('author')
    }
    books.append(new_book)
    return jsonify(new_book), 201


@app.route('/books/<int:book_id>', methods=['PUT'])
def update_book(book_id):
    book = next((book for book in books if book['id'] == book_id), None)
    if book is None:
        return jsonify({'message': 'Book not found'}), 404
    data = request.get_json()
    book['title'] = data.get('title', book['title'])
    book['author'] = data.get('author', book['author'])
    return jsonify(book)


@app.route('/books/<int:book_id>', methods=['DELETE'])
def delete_book(book_id):
    book = next((book for book in books if book['id'] == book_id), None)
    if book is None:
        return jsonify({'message': 'Book not found'}), 404
    books.remove(book)
    return jsonify({'message': 'Book deleted'})


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

在这段代码中：

• @app.route('/books', methods=['GET']) 定义了获取所有书籍的端点，使用 jsonify 将书籍列表转换为 JSON 格式返回。
• @app.route('/books/<int:book_id>', methods=['GET']) 用于获取特定 ID 的书籍，通过 next 函数查找书籍，如果未找到返回 404 错误。
• @app.route('/books', methods=['POST']) 处理创建新书籍的请求，从请求中获取 JSON 数据并添加到书籍列表中，返回创建的新书籍并附带 201 状态码。
• @app.route('/books/<int:book_id>', methods=['PUT']) 用于更新特定书籍，同样先查找书籍，然后根据请求数据更新书籍信息。
• @app.route('/books/<int:book_id>', methods=['DELETE']) 处理删除特定书籍的请求，找到书籍后从列表中移除。

四、使用 Django 构建 RESTful API

1. 安装 Django 使用 pip 安装 Django：

pip install django

2. 创建 Django 项目 在命令行中执行以下命令创建一个新的 Django 项目：

django - admin startproject myproject
cd myproject

3. 创建应用 在项目目录下创建一个新的应用，例如 books：

python manage.py startapp books

4. 定义模型 在 books/models.py 文件中定义书籍模型：

from django.db import models


class Book(models.Model):
    title = models.CharField(max_length=200)
    author = models.CharField(max_length=200)


    def __str__(self):
        return self.title

5. 创建数据库表 在项目目录下执行以下命令来创建数据库表：

python manage.py makemigrations
python manage.py migrate

6. 创建序列化器 为了将模型实例转换为 JSON 格式并反之，我们需要使用 Django REST framework。首先安装它：

pip install djangorestframework

在 books 应用下创建 serializers.py 文件：

from rest_framework import serializers
from.models import Book


class BookSerializer(serializers.ModelSerializer):
    class Meta:
        model = Book
        fields = '__all__'

7. 定义视图 在 books/views.py 文件中定义视图：

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from.models import Book
from.serializers import BookSerializer


class BookList(APIView):
    def get(self, request):
        books = Book.objects.all()
        serializer = BookSerializer(books, many=True)
        return Response(serializer.data)

    def post(self, request):
        serializer = BookSerializer(data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)


class BookDetail(APIView):
    def get_object(self, pk):
        try:
            return Book.objects.get(pk=pk)
        except Book.DoesNotExist:
            raise Http404

    def get(self, request, pk):
        book = self.get_object(pk)
        serializer = BookSerializer(book)
        return Response(serializer.data)

    def put(self, request, pk):
        book = self.get_object(pk)
        serializer = BookSerializer(book, data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    def delete(self, request, pk):
        book = self.get_object(pk)
        book.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

8. 配置路由 在 myproject/urls.py 文件中配置路由：

from django.contrib import admin
from django.urls import path, include


urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/books/', include('books.urls')),
]

在 books/urls.py 文件中定义书籍相关的路由：

from django.urls import path
from. import views


urlpatterns = [
    path('', views.BookList.as_view(), name='book - list'),
    path('<int:pk>/', views.BookDetail.as_view(), name='book - detail'),
]

通过以上步骤，我们使用 Django 和 Django REST framework 构建了一个简单的书籍管理 RESTful API。

五、使用 FastAPI 构建 RESTful API

1. 安装 FastAPI 和 Uvicorn FastAPI 通常与 Uvicorn 一起使用，Uvicorn 是一个基于 ASGI 的高性能服务器。安装命令如下：

pip install fastapi uvicorn

2. 创建基本的 FastAPI 应用 创建一个新的 Python 文件，例如 main.py：

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
def read_root():
    return {'Hello': 'World'}

在上述代码中，我们导入 FastAPI 类并创建了一个实例 app。通过 @app.get 装饰器定义了一个根路由，返回一个包含“Hello: World”的 JSON 数据。

3. 实现 RESTful API 端点 假设我们还是构建一个书籍管理 API，代码如下：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class Book(BaseModel):
    id: int
    title: str
    author: str


books = [
    Book(id=1, title='Python Crash Course', author='Eric Matthes'),
    Book(id=2, title='Clean Code', author='Robert C. Martin')
]


@app.get('/books')
def get_books():
    return books


@app.get('/books/{book_id}')
def get_book(book_id: int):
    book = next((book for book in books if book.id == book_id), None)
    if book is None:
        raise HTTPException(status_code=404, detail='Book not found')
    return book


@app.post('/books')
def create_book(book: Book):
    new_id = max(book.id for book in books) + 1 if books else 1
    new_book = Book(id=new_id, title=book.title, author=book.author)
    books.append(new_book)
    return new_book


@app.put('/books/{book_id}')
def update_book(book_id: int, book: Book):
    existing_book = next((existing for existing in books if existing.id == book_id), None)
    if existing_book is None:
        raise HTTPException(status_code=404, detail='Book not found')
    existing_book.title = book.title
    existing_book.author = book.author
    return existing_book


@app.delete('/books/{book_id}')
def delete_book(book_id: int):
    book = next((book for book in books if book.id == book_id), None)
    if book is None:
        raise HTTPException(status_code=404, detail='Book not found')
    books.remove(book)
    return {'message': 'Book deleted'}

在这段代码中：

• 我们首先定义了一个 Book 类，它继承自 BaseModel，用于数据验证和序列化。
• @app.get('/books') 获取所有书籍列表。
• @app.get('/books/{book_id}') 获取特定 ID 的书籍，若未找到抛出 404 异常。
• @app.post('/books') 创建新书籍，为新书籍分配一个新的 ID 并添加到列表中。
• @app.put('/books/{book_id}') 更新特定书籍，找到书籍后更新其属性。
• @app.delete('/books/{book_id}') 删除特定书籍，若未找到同样抛出 404 异常。

六、认证与授权

1. 认证 认证是验证客户端身份的过程。在 RESTful API 中，常见的认证方式有：

• 基本认证：客户端在每个请求的 Authorization 头中发送用户名和密码的 Base64 编码字符串。在 Flask 中可以使用 flask_httpauth 扩展来实现基本认证：

from flask import Flask
from flask_httpauth import HTTPBasicAuth

app = Flask(__name__)
auth = HTTPBasicAuth()

users = {
    "admin": "password"
}

@auth.verify_password
def verify_password(username, password):
    if username in users and users[username] == password:
        return True
    return False

@app.route('/protected')
@auth.login_required
def protected():
    return "This is a protected resource"

• 令牌认证：客户端在登录成功后会收到一个令牌（token），后续请求将令牌放在 Authorization 头中。在 Django REST framework 中，可以使用 rest_framework_simplejwt 来实现令牌认证。首先安装：

pip install djangorestframework - simplejwt

然后在 settings.py 中配置：

from datetime import timedelta

SIMPLE_JWT = {
    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=15),
    'REFRESH_TOKEN_LIFETIME': timedelta(days=1),
}

在视图中使用认证：

from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework.views import APIView
from rest_framework.response import Response


class ProtectedView(APIView):
    authentication_classes = [JWTAuthentication]

    def get(self, request):
        return Response({'message': 'This is a protected resource'})

2. 授权 授权是确定已认证的客户端是否有权限执行特定操作的过程。例如，只有管理员用户才能删除书籍。在 Flask 中，可以在视图函数中添加权限检查逻辑：

from flask import Flask
from flask_httpauth import HTTPBasicAuth

app = Flask(__name__)
auth = HTTPBasicAuth()

users = {
    "admin": "password",
    "user": "userpass"
}

@auth.verify_password
def verify_password(username, password):
    if username in users and users[username] == password:
        return True
    return False

@app.route('/books/<int:book_id>', methods=['DELETE'])
@auth.login_required
def delete_book(book_id):
    if auth.username()!= 'admin':
        return "You don't have permission to delete this book", 403
    # 执行删除书籍逻辑
    return "Book deleted"

在 Django REST framework 中，可以使用权限类来实现授权。例如，创建一个自定义权限类：

from rest_framework.permissions import BasePermission


class IsAdmin(BasePermission):
    def has_permission(self, request, view):
        return request.user.is_staff

然后在视图中使用该权限类：

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from. import permissions


class DeleteBookView(APIView):
    permission_classes = [IsAuthenticated, permissions.IsAdmin]

    def delete(self, request, book_id):
        # 执行删除书籍逻辑
        return Response({'message': 'Book deleted'})

七、错误处理与日志记录

1. 错误处理 在 Flask 中，可以使用 @app.errorhandler 装饰器来处理全局错误。例如，处理 404 错误：

from flask import Flask, jsonify

app = Flask(__name__)


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

在 Django REST framework 中，默认已经有较好的错误处理机制。但也可以自定义异常处理，例如：

from rest_framework.views import exception_handler
from rest_framework.response import Response


def custom_exception_handler(exc, context):
    response = exception_handler(exc, context)
    if response is None:
        return Response({'message': 'An unexpected error occurred'}, status=500)
    return response

然后在 settings.py 中配置自定义异常处理：

REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'your_app_name.utils.custom_exception_handler'
}

在 FastAPI 中，可以使用 @app.exception_handler 来处理特定异常。例如，处理 HTTPException：

from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.exception_handler(HTTPException)
def http_exception_handler(request, exc):
    return {'message': exc.detail}, exc.status_code

2. 日志记录 在 Python 中，内置的 logging 模块可以用于日志记录。在 Flask 应用中，可以这样配置日志：

import logging
from flask import Flask

app = Flask(__name__)

logging.basicConfig(filename='app.log', level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')


@app.route('/')
def hello_world():
    try:
        result = 1 / 0
    except ZeroDivisionError as e:
        logging.error(f'Error occurred: {str(e)}')
    return 'Hello, World!'

在 Django 中，可以在 settings.py 中配置日志：

LOGGING = {
   'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'file': {
            'level': 'DEBUG',
            'class': 'logging.FileHandler',
            'filename': 'django.log',
        },
    },
    'loggers': {
        'django': {
            'handlers': ['file'],
            'level': 'DEBUG',
            'propagate': True,
        },
    },
}

在 FastAPI 中，同样可以使用 logging 模块：

import logging
from fastapi import FastAPI

app = FastAPI()

logging.basicConfig(filename='fastapi.log', level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')


@app.get('/')
def read_root():
    logging.info('Accessed root endpoint')
    return {'Hello': 'World'}

通过合理的错误处理和日志记录，可以提高 RESTful API 的稳定性和可维护性。

八、性能优化

1. 缓存 缓存可以显著提高 API 的性能，减少数据库查询次数。在 Flask 中，可以使用 Flask - Caching 扩展。首先安装：

pip install Flask - Caching

然后在应用中配置：

from flask import Flask
from flask_caching import Cache

app = Flask(__name__)
cache = Cache(app, config={'CACHE_TYPE':'simple'})


@app.route('/books')
@cache.cached(timeout=60)
def get_books():
    # 从数据库获取书籍列表逻辑
    return jsonify(books)

在 Django 中，内置了缓存支持。可以在 settings.py 中配置缓存：

CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
        'LOCATION': 'unique - cache - location'
    }
}

在视图中使用缓存：

from django.views.decorators.cache import cache_page
from django.http import HttpResponse
from.models import Book


@cache_page(60 * 15)  # 缓存 15 分钟
def book_list(request):
    books = Book.objects.all()
    # 渲染书籍列表视图
    return HttpResponse('Book list')

在 FastAPI 中，可以使用 cachetools 库实现缓存：

from fastapi import FastAPI
from cachetools import cached, TTLCache

app = FastAPI()
cache = TTLCache(maxsize=100, ttl=60)


@app.get('/books')
@cached(cache)
def get_books():
    # 获取书籍列表逻辑
    return books

2. 数据库优化

• 索引：在 Django 中，可以在模型字段上添加索引来提高查询性能。例如：

from django.db import models


class Book(models.Model):
    title = models.CharField(max_length=200, db_index=True)
    author = models.CharField(max_length=200)

• 批量操作：在处理数据库操作时，尽量使用批量操作。例如，在 Django 中批量创建书籍：

books_data = [
    {'title': 'Book 1', 'author': 'Author 1'},
    {'title': 'Book 2', 'author': 'Author 2'}
]
books = [Book(**data) for data in books_data]
Book.objects.bulk_create(books)

3. 异步处理 FastAPI 基于 ASGI 支持异步编程，可以提高应用在高并发场景下的性能。例如，定义一个异步视图：

import asyncio
from fastapi import FastAPI

app = FastAPI()


@app.get('/async - task')
async def async_task():
    await asyncio.sleep(1)
    return {'message': 'Async task completed'}

通过这些性能优化手段，可以让 RESTful API 在处理大量请求时更加高效稳定。