RestFul风格代码库

以drf代码为参考基线, 路由命名参考flask文档的说明: https://flask.palletsprojects.com/en/2.3.x/views/#method-dispatching-and-apis

技术栈:

完整版: 在bilibili-api各框架下examples案例的project目录下

项目工程结构

project/app ; go-zero创建的api服务
        model ; gorm模型
        scripts/ddl ; sql脚本
                gencode/configuration.go ; 生成代码脚本
        tests  ; 接口测试
        go.mod

模型

from django.db import models

# 从自动生成的模型文件导入模型
from .gen import Author


# 在代码声明模型
# =========================================
class Story(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    author = models.ForeignKey(Author, on_delete=models.CASCADE)

    def __str__(self):
        return self.title

视图

application/views.py

列表查询

简单模型查询

app/api/routes/users.py

from sqlalchemy import func, select

@router.get(
    "/",
    dependencies=[Depends(get_current_active_superuser)],
    response_model=UsersPublic,
)
def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
    """Retrieve users."""
    count_statement = select(func.count()).select_from(User)
    count = session.execute(count_statement).scalar_one()

    statement = select(User).offset(skip).limit(limit)
    users = session.execute(statement).scalars().all()

    return UsersPublic(data=users, count=count)

指定返回字段，在多字段模型下能提高查询性能

app/api/routes/items.py

from sqlalchemy import func, select
from sqlalchemy.orm import load_only

@router.get("/", response_model=ItemsPublic)
def read_items(
    session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
) -> Any:
    """获取项目列表"""
    if current_user.is_superuser:
        count_statement = select(func.count()).select_from(Item)
        # 手工指定字段数据返回
        statement = select(
            Item.id,
            Item.owner_id,
            Item.title,
            Item.description
        ).offset(skip).limit(limit)
        items: list[dict] = session.execute(statement).mappings().all()
    else:
        count_statement = (
            select(func.count())
            .select_from(Item)
            .where(Item.owner_id == current_user.id)
        )
        # 方式二: https://docs.sqlalchemy.org/en/20/orm/queryguide/columns.html#using-load-only-to-reduce-loaded-columns
        statement = (
            select(Item)
            .options(load_only(
                Item.id,
                Item.owner_id,
                Item.title,
                Item.description
            ))
            .where(Item.owner_id == current_user.id)
            .offset(skip)
            .limit(limit)
        )
        items: list[Item] = session.execute(statement).scalars().all()

    count = session.execute(count_statement).scalar_one()
    return ItemsPublic(data=items, count=count)

路由

project/router.py

from rest_framework import routers

from application.views import StoryViewSet, StoryV2ViewSet, StoryV3ViewSet

# https://www.django-rest-framework.org/api-guide/routers/#usage
router = routers.SimpleRouter()
# 自动生成url名字可在reverse函数中使用
# * story-detail: get(单个)、patch、put、delete请求
# * story-list: get(列表)、post请求
router.register('v1/stories', StoryViewSet)
router.register('v2/stories', StoryV2ViewSet, basename='storyV2')
router.register('v3/stories', StoryV3ViewSet, basename='storyV3')

测试

application/tests.py

from django.urls import reverse
from rest_framework.test import APITestCase

from application.models import Author


class StoryTests(APITestCase):
    @classmethod
    def setUpTestData(cls) -> None:
        cls.author = Author(id=1, name='Beatles', age=18)
        cls.author.save()

    def test_create_story(self):
        """
        Ensure we can create a new story.
        """
        for name, view_name in [
            ('全自动视图', 'story-list'),
            ('半自动视图', 'storyV2-list'),
            ('全手动视图', 'storyV3-list')
        ]:
            url = reverse(view_name)
            with self.subTest(name=name):
                data = {'title': 'Test Story',
                        'content': 'This is a test story.',
                        'author': self.author.pk}
                response = self.client.post(url, data, format='json')
                self.assertEqual(response.status_code, 201)
                self.assertEqual(response.data['title'], 'Test Story')
                self.assertEqual(response.data['author'], self.author.pk)

小技巧

create接口的测试思路参考drf的 example

  • 验证状态码

  • 用模型查询表的数量 -> 从返回值获取id， 然后用模型查id

  • 获取表的第一条数据，验证名称

swagger