当前位置：首页 > news >正文

用Django REST Framework从零搭建共享充电桩后台API（附完整项目结构）

news 2026/3/31 6:46:27

从零构建共享充电桩API：Django 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 django==3.2.16 djangorestframework==3.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_length=32, unique=True) location = models.CharField(max_length=200) power_output = models.PositiveIntegerField( # 输出功率(W) validators=[MinValueValidator(1000), MaxValueValidator(22000)] ) status = models.CharField( max_length=12, choices=Status.choices, default=Status.ONLINE ) last_heartbeat = models.DateTimeField(null=True) 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_delete=models.PROTECT) charger = models.ForeignKey('chargers.Charger', on_delete=models.PROTECT) start_time = models.DateTimeField(auto_now_add=True) end_time = models.DateTimeField(null=True) consumed_kwh = models.DecimalField( max_digits=6, decimal_places=2, validators=[MinValueValidator(0.01)], null=True ) status = models.CharField( max_length=10, choices=Status.choices, default=Status.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(detail=True, methods=['post']) def heartbeat(self, request, pk=None): charger = self.get_object() charger.last_heartbeat = timezone.now() charger.save() return Response({'status': 'updated'}) @action(detail=True, methods=['post']) def start_charging(self, request, pk=None): # 实现启动充电的业务逻辑 ...

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( user=serializer.validated_data['user'], status__in=[Order.Status.CREATED, Order.Status.CHARGING] ).exists(): raise ValidationError('存在进行中的订单') serializer.save(status=Order.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.py：

INSTALLED_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(), name='schema'), path('api/docs/', SpectacularSwaggerView.as_view(url_name='schema'), name='docs'), ]

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_id="TEST001", location="测试位置", power_output=7500 ) 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.py：

from .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(f"API异常: {str(exception)}", exc_info=True) return Response( {'error': '服务器内部错误'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR )

查看全文

http://www.jsqmd.com/news/563976/