从零构建共享充电桩APIDjango REST Framework工程化实践共享充电桩作为城市基础设施的重要组成部分其后台系统的稳定性和扩展性直接影响用户体验。本文将带你从零开始用Django REST Framework构建一个符合生产标准的API系统。不同于简单的功能堆砌我们更关注项目脚手架设计、代码组织规范和API设计原则这些才是支撑长期迭代的关键。1. 环境配置与项目初始化在开始编码前合理的开发环境配置能避免后续大量坑点。推荐使用Python 3.8和Django 3.2的组合这是目前企业级项目最稳定的版本搭配。# 创建虚拟环境推荐使用venv python -m venv venv source venv/bin/activate # Linux/Mac venv\Scripts\activate # Windows # 安装核心依赖 pip install django3.2.16 djangorestframework3.14.0项目结构设计应该遵循模块化和关注点分离原则。以下是经过多个物联网项目验证的标准结构shared_charging/ ├── config/ # 项目配置目录原settings.py位置 │ ├── __init__.py │ ├── settings.py # 拆分为多个配置文件 │ ├── urls.py │ └── asgi.py ├── apps/ # 所有应用模块 │ ├── chargers/ # 充电桩模块 │ ├── users/ # 用户模块 │ └── ... # 其他模块 ├── scripts/ # 部署和管理脚本 ├── requirements/ # 分环境依赖文件 │ ├── base.txt # 基础依赖 │ └── local.txt # 开发环境特有依赖 └── manage.py提示使用python manage.py startapp创建应用时建议通过--template参数加载自定义模板确保所有应用初始结构一致。2. 数据模型设计实战充电桩系统的核心在于数据模型的设计。我们需要考虑设备状态管理、订单生命周期和计费策略等多个业务维度。2.1 充电桩模型设计# apps/chargers/models.py from django.db import models from django.core.validators import MinValueValidator, MaxValueValidator class Charger(models.Model): class Status(models.TextChoices): ONLINE online, 在线 OFFLINE offline, 离线 MAINTENANCE maintenance, 维护中 charger_id models.CharField(max_length32, uniqueTrue) location models.CharField(max_length200) power_output models.PositiveIntegerField( # 输出功率(W) validators[MinValueValidator(1000), MaxValueValidator(22000)] ) status models.CharField( max_length12, choicesStatus.choices, defaultStatus.ONLINE ) last_heartbeat models.DateTimeField(nullTrue) def __str__(self): return f{self.charger_id}{self.location}2.2 订单模型设计订单系统需要处理状态流转和计费逻辑# apps/orders/models.py from django.db import models from django.core.validators import MinValueValidator class Order(models.Model): class Status(models.TextChoices): CREATED created, 已创建 CHARGING charging, 充电中 COMPLETED completed, 已完成 CANCELLED cancelled, 已取消 user models.ForeignKey(users.User, on_deletemodels.PROTECT) charger models.ForeignKey(chargers.Charger, on_deletemodels.PROTECT) start_time models.DateTimeField(auto_now_addTrue) end_time models.DateTimeField(nullTrue) consumed_kwh models.DecimalField( max_digits6, decimal_places2, validators[MinValueValidator(0.01)], nullTrue ) status models.CharField( max_length10, choicesStatus.choices, defaultStatus.CREATED ) property def duration(self): if self.end_time: return self.end_time - self.start_time return None注意金额计算建议使用DecimalField而非FloatField避免浮点数精度问题。3. API端点设计与实现RESTful API设计应该遵循资源导向原则同时考虑物联网设备的特殊性。3.1 充电桩API实现使用DRF的ViewSet可以快速构建CRUD接口但需要添加设备状态管理等业务逻辑# apps/chargers/api/views.py from rest_framework import viewsets, status from rest_framework.decorators import action from rest_framework.response import Response from ..models import Charger from .serializers import ChargerSerializer class ChargerViewSet(viewsets.ModelViewSet): queryset Charger.objects.all() serializer_class ChargerSerializer action(detailTrue, methods[post]) def heartbeat(self, request, pkNone): charger self.get_object() charger.last_heartbeat timezone.now() charger.save() return Response({status: updated}) action(detailTrue, methods[post]) def start_charging(self, request, pkNone): # 实现启动充电的业务逻辑 ...3.2 订单API的特殊处理订单创建需要处理业务验证和状态机转换# apps/orders/api/views.py from rest_framework import viewsets from rest_framework.exceptions import ValidationError from ..models import Order class OrderViewSet(viewsets.ModelViewSet): queryset Order.objects.all() serializer_class OrderSerializer def perform_create(self, serializer): charger serializer.validated_data[charger] if charger.status ! Charger.Status.ONLINE: raise ValidationError(充电桩不可用) # 检查用户是否有未完成订单 if Order.objects.filter( userserializer.validated_data[user], status__in[Order.Status.CREATED, Order.Status.CHARGING] ).exists(): raise ValidationError(存在进行中的订单) serializer.save(statusOrder.Status.CREATED)4. 项目进阶配置4.1 分页与过滤配置在config/settings.py中添加DRF全局配置REST_FRAMEWORK { DEFAULT_PAGINATION_CLASS: rest_framework.pagination.PageNumberPagination, PAGE_SIZE: 20, DEFAULT_FILTER_BACKENDS: [ django_filters.rest_framework.DjangoFilterBackend, rest_framework.filters.OrderingFilter ], }4.2 API文档生成使用drf-yasg或drf-spectacular自动生成API文档pip install drf-spectacular配置settings.pyINSTALLED_APPS [drf_spectacular] REST_FRAMEWORK { # ...其他配置 DEFAULT_SCHEMA_CLASS: drf_spectacular.openapi.AutoSchema, } SPECTACULAR_SETTINGS { TITLE: 共享充电桩API文档, VERSION: 1.0.0, SERVE_INCLUDE_SCHEMA: False, }在根urls.py中添加from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView urlpatterns [ # ...其他路由 path(api/schema/, SpectacularAPIView.as_view(), nameschema), path(api/docs/, SpectacularSwaggerView.as_view(url_nameschema), namedocs), ]5. 测试与部署准备5.1 编写API测试用例使用DRF的APITestCase编写集成测试# apps/chargers/tests/test_api.py from rest_framework.test import APITestCase from django.urls import reverse from ..models import Charger class ChargerAPITests(APITestCase): classmethod def setUpTestData(cls): cls.charger Charger.objects.create( charger_idTEST001, location测试位置, power_output7500 ) def test_get_charger_list(self): url reverse(charger-list) response self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data[results]), 1)5.2 生产环境配置建议创建config/settings/production.pyfrom .base import * DEBUG False DATABASES { default: { ENGINE: django.db.backends.postgresql, NAME: os.getenv(DB_NAME), USER: os.getenv(DB_USER), PASSWORD: os.getenv(DB_PASSWORD), HOST: os.getenv(DB_HOST), PORT: os.getenv(DB_PORT, 5432), } } # 安全配置 SECURE_HSTS_SECONDS 3600 SECURE_SSL_REDIRECT True SESSION_COOKIE_SECURE True CSRF_COOKIE_SECURE True在项目实践中我们发现合理的错误处理能大幅降低运维成本。建议为所有API添加统一的异常处理中间件# config/middleware/exception_handler.py import logging from rest_framework.response import Response from rest_framework import status logger logging.getLogger(__name__) class APIExceptionHandler: def __init__(self, get_response): self.get_response get_response def __call__(self, request): return self.get_response(request) def process_exception(self, request, exception): logger.error(fAPI异常: {str(exception)}, exc_infoTrue) return Response( {error: 服务器内部错误}, statusstatus.HTTP_500_INTERNAL_SERVER_ERROR )