Django+MySQL接口开发完全指南

前言

本文将详细介绍如何使用Django结合MySQL数据库开发RESTful API接口。我们将从环境搭建开始，一步步实现一个完整的接口项目。

环境准备

首先需要安装以下组件：

• Python 3.8+
• Django 4.2
• MySQL 8.0
• mysqlclient
• djangorestframework

安装命令

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows使用 venv\Scripts\activate

# 安装依赖
pip install django==4.2
pip install mysqlclient
pip install djangorestframework

项目创建与配置

1. 创建项目

django-admin startproject myproject
cd myproject
python manage.py startapp api

2. 配置settings.py

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',  # 添加rest_framework
    'api',  # 添加api应用
]

# 数据库配置
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'django_demo',
        'USER': 'root',
        'PASSWORD': 'your_password',
        'HOST': '127.0.0.1',
        'PORT': '3306',
    }
}

# REST框架配置
REST_FRAMEWORK = {
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 10,
}

数据库模型设计

1. 创建模型（api/models.py）

from django.db import models

class Book(models.Model):
    title = models.CharField(max_length=100, verbose_name='书名')
    author = models.CharField(max_length=50, verbose_name='作者')
    price = models.DecimalField(max_digits=5, decimal_places=2, verbose_name='价格')
    publish_date = models.DateField(verbose_name='出版日期')
    created_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    updated_at = models.DateTimeField(auto_now=True, verbose_name='更新时间')

    class Meta:
        db_table = 'books'
        verbose_name = '图书'
        verbose_name_plural = verbose_name
        ordering = ['-created_at']

    def __str__(self):
        return self.title

2. 生成并应用数据库迁移

python manage.py makemigrations
python manage.py migrate

序列化器开发

1. 创建序列化器（api/serializers.py）

from rest_framework import serializers
from .models import Book

class BookSerializer(serializers.ModelSerializer):
    class Meta:
        model = Book
        fields = ['id', 'title', 'author', 'price', 'publish_date', 'created_at', 'updated_at']
        read_only_fields = ['created_at', 'updated_at']

视图开发

1. 创建视图（api/views.py）

from rest_framework import viewsets, filters
from rest_framework.response import Response
from rest_framework.decorators import action
from django_filters.rest_framework import DjangoFilterBackend
from .models import Book
from .serializers import BookSerializer

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_fields = ['author']
    search_fields = ['title', 'author']
    ordering_fields = ['price', 'publish_date']

    @action(detail=False, methods=['get'])
    def get_book_stats(self, request):
        """获取图书统计信息"""
        total_books = Book.objects.count()
        total_authors = Book.objects.values('author').distinct().count()
        avg_price = Book.objects.aggregate(avg_price=models.Avg('price'))

        return Response({
            'total_books': total_books,
            'total_authors': total_authors,
            'average_price': avg_price['avg_price']
        })

URL配置

1. 配置项目URLs（myproject/urls.py）

from django.contrib import admin
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from api.views import BookViewSet

router = DefaultRouter()
router.register(r'books', BookViewSet)

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

中间件开发

1. 创建自定义中间件（api/middleware.py）

import time
from django.utils.deprecation import MiddlewareMixin
import logging

logger = logging.getLogger(__name__)

class RequestLogMiddleware(MiddlewareMixin):
    def process_request(self, request):
        request.start_time = time.time()

    def process_response(self, request, response):
        if hasattr(request, 'start_time'):
            duration = time.time() - request.start_time
            logger.info(f'{request.method} {request.path} - {response.status_code} - {duration:.2f}s')
        return response

2. 在settings.py中注册中间件

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    'api.middleware.RequestLogMiddleware',  # 添加自定义中间件
]

异常处理

1. 创建自定义异常处理（api/exceptions.py）

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

def custom_exception_handler(exc, context):
    response = exception_handler(exc, context)

    if response is None:
        return Response({
            'error': str(exc),
            'message': '服务器内部错误'
        }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

    return response

2. 在settings.py中配置异常处理器

REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'api.exceptions.custom_exception_handler'
}

API接口测试

1. 创建测试用例（api/tests.py）

from rest_framework.test import APITestCase
from rest_framework import status
from .models import Book
from datetime import date

class BookTests(APITestCase):
    def setUp(self):
        self.book_data = {
            'title': 'Django实战',
            'author': '张三',
            'price': '59.99',
            'publish_date': '2024-01-01'
        }
        self.book = Book.objects.create(**self.book_data)

    def test_create_book(self):
        """测试创建图书"""
        response = self.client.post('/api/books/', self.book_data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertEqual(Book.objects.count(), 2)

    def test_get_book_list(self):
        """测试获取图书列表"""
        response = self.client.get('/api/books/')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data['results']), 1)

API接口文档

使用drf-yasg生成Swagger文档：

1. 安装drf-yasg

pip install drf-yasg

2. 配置settings.py

INSTALLED_APPS = [
    ...
    'drf_yasg',
]

3. 配置URLs（myproject/urls.py）

from django.urls import path, include, re_path
from rest_framework import permissions
from drf_yasg.views import get_schema_view
from drf_yasg import openapi

schema_view = get_schema_view(
   openapi.Info(
      title="图书管理API",
      default_version='v1',
      description="图书管理系统API文档",
      terms_of_service="https://www.yourapp.com/terms/",
      contact=openapi.Contact(email="[email protected]"),
      license=openapi.License(name="BSD License"),
   ),
   public=True,
   permission_classes=(permissions.AllowAny,),
)

urlpatterns = [
    ...
    re_path(r'^swagger(?P<format>\.json|\.yaml)$', schema_view.without_ui(cache_timeout=0), name='schema-json'),
    path('swagger/', schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'),
    path('redoc/', schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'),
]

接口使用示例

1. 创建图书

curl -X POST http://localhost:8000/api/books/ \
     -H "Content-Type: application/json" \
     -d '{"title":"Django实战","author":"张三","price":"59.99","publish_date":"2024-01-01"}'

2. 获取图书列表

curl http://localhost:8000/api/books/

3. 更新图书

curl -X PUT http://localhost:8000/api/books/1/ \
     -H "Content-Type: application/json" \
     -d '{"title":"Django实战（第二版）","author":"张三","price":"69.99","publish_date":"2024-01-01"}'

4. 删除图书

curl -X DELETE http://localhost:8000/api/books/1/

性能优化建议

1. 使用数据库索引

   class Book(models.Model):
       ...
       class Meta:
           indexes = [
               models.Index(fields=['title']),
               models.Index(fields=['author']),
           ]

2. 使用缓存

   from django.core.cache import cache
   
   class BookViewSet(viewsets.ModelViewSet):
       def list(self, request):
           cache_key = 'book_list'
           cached_data = cache.get(cache_key)
           
           if cached_data is None:
               queryset = self.filter_queryset(self.get_queryset())
               serializer = self.get_serializer(queryset, many=True)
               cached_data = serializer.data
               cache.set(cache_key, cached_data, timeout=300)  # 缓存5分钟
               
           return Response(cached_data)

总结

本文详细介绍了使用Django开发RESTful API的完整流程，包括：

1. 环境搭建和项目配置
2. 数据库模型设计
3. 序列化器开发
4. 视图和URL配置
5. 中间件和异常处理
6. 测试用例编写
7. API文档生成
8. 性能优化建议

通过按照本教程的步骤，你可以快速搭建一个功能完善的Django API项目。建议在实际开发中根据具体需求进行适当调整和扩展。