Django 业务层 service 基类封装

2023-04-25 03:00:28 ⋅ 5319 ⋅ 0 ⋅ 0

需求

Django业务层写一个service基类，其他业务 service继承该类，做实际业务逻辑开发

实现

方案一

在Django中，业务层通常用于处理复杂的业务逻辑和数据操作。为了实现代码的复用和扩展性，可以创建一个Service基类，然后让其他业务Service类继承它。以下是一个示例：

1、首先，在你的Django项目中创建一个名为services的文件夹，然后在该文件夹中创建一个名为base_service.py的文件。这个文件将包含我们的基类。
2、编写基类base_service.py：

from django.core.exceptions import ObjectDoesNotExist

class BaseService:
    model = None

    def __init__(self):
        if not self.model:
            raise ValueError("model属性必须在子类中定义")

    def get(self, *args, **kwargs):
        try:
            instance = self.model.objects.get(*args, **kwargs)
        except ObjectDoesNotExist:
            instance = None
        return instance

    def filter(self, *args, **kwargs):
        return self.model.objects.filter(*args, **kwargs)

    def create(self, **kwargs):
        return self.model.objects.create(**kwargs)

    def update(self, instance, **kwargs):
        for attr, value in kwargs.items():
            setattr(instance, attr, value)
        instance.save()
        return instance

    def delete(self, instance):
        instance.delete()

    # 可以根据需要添加其他通用方法

创建一个具体的业务Service类，例如UserService，让它继承BaseService类，并在其中实现具体的业务逻辑。例如，在services文件夹下创建user_service.py：

from .base_service import BaseService
from your_app.models import User  # 从你的应用中导入User模型

class UserService(BaseService):
    model = User

    def __init__(self):
        super().__init__()

    # 在这里实现与User相关的业务逻辑
    # 例如：创建用户、更新用户信息、删除用户等

    def get_users_by_age(self, age):
        return self.filter(age=age)

在视图中使用UserService：

from django.http import JsonResponse
from services.user_service import UserService

def get_users_by_age(request, age):
    user_service = UserService()
    users = user_service.get_users_by_age(age)
    user_list = []
    for user in users:
        user_list.append({"id": user.id, "username": user.username, "age": user.age})
    return JsonResponse({"users": user_list})

这样，我们就创建了一个基本的Service基类和一个具体的UserService类。可以根据实际需求继续在UserService中添加业务逻辑。

方案二：静态方法实现

请用静态方法进行重构优化，不需要创建对象
可以将BaseService中的方法修改为静态方法，并在子类中通过类名直接调用。这样就不需要创建类的实例。以下是修改后的BaseService和UserService类：

  1. 修改base_service.py：
from django.core.exceptions import ObjectDoesNotExist

class BaseService:
    model = None

    @staticmethod
    def get(model, *args, **kwargs):
        try:
            instance = model.objects.get(*args, **kwargs)
        except ObjectDoesNotExist:
            instance = None
        return instance

    @staticmethod
    def filter(model, *args, **kwargs):
        return model.objects.filter(*args, **kwargs)

    @staticmethod
    def create(model, **kwargs):
        return model.objects.create(**kwargs)

    @staticmethod
    def update(instance, **kwargs):
        for attr, value in kwargs.items():
            setattr(instance, attr, value)
        instance.save()
        return instance

    @staticmethod
    def delete(instance):
        instance.delete()

    # 可以根据需要添加其他通用方法
  1. 修改user_service.py：
from .base_service import BaseService
from your_app.models import User  # 从你的应用中导入User模型

class UserService(BaseService):

    # 在这里实现与User相关的业务逻辑
    # 例如：创建用户、更新用户信息、删除用户等

    @staticmethod
    def get_users_by_age(age):
        return BaseService.filter(User, age=age)
  1. 在视图中使用UserService：
from django.http import JsonResponse
from services.user_service import UserService

def get_users_by_age(request, age):
    users = UserService.get_users_by_age(age)
    user_list = []
    for user in users:
        user_list.append({"id": user.id, "username": user.username, "age": user.age})
    return JsonResponse({"users": user_list})

通过这种方式，您可以在不创建对象的情况下使用静态方法。但请注意，这将限制了类的灵活性，因为静态方法无法访问类的实例属性。在这个示例中，我们不再需要子类的__init__方法。

方案三、继续优化

请根据项目实战，继续优化上述基类及实战示例

以下是根据项目实战需求，对基类和实战示例进行优化的建议：

  1. 在BaseService类中添加异常处理，使代码更健壮。

  2. 使用Python的@classmethod装饰器，这样子类可以更好地继承和重写基类的方法。

  3. 为了提高代码的可读性，使用类型提示（Type Hints）。

修改后的base_service.py：

from typing import Type, TypeVar, List, Optional
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import Model, QuerySet

T = TypeVar("T", bound=Model)

class BaseService:
    model: Type[T]

    @classmethod
    def get(cls, *args, **kwargs) -> Optional[T]:
        try:
            instance = cls.model.objects.get(*args, **kwargs)
        except ObjectDoesNotExist:
            instance = None
        return instance

    @classmethod
    def filter(cls, *args, **kwargs) -> QuerySet[T]:
        return cls.model.objects.filter(*args, **kwargs)

    @classmethod
    def create(cls, **kwargs) -> T:
        return cls.model.objects.create(**kwargs)

    @classmethod
    def update(cls, instance: T, **kwargs) -> T:
        for attr, value in kwargs.items():
            setattr(instance, attr, value)
        instance.save()
        return instance

    @classmethod
    def delete(cls, instance: T) -> None:
        instance.delete()

    # 可以根据需要添加其他通用方法

修改后的user_service.py：

更新用户信息、删除用户等

修改后的user_service.py：

from typing import List
from .base_service import BaseService, T
from your_app.models import User  # 从你的应用中导入User模型

class UserService(BaseService):
    model = User

    # 在这里实现与User相关的业务逻辑
    # 例如：创建用户、更新用户信息、删除用户等

    @classmethod
    def get_users_by_age(cls, age: int) -> List[T]:
        return cls.filter(age=age)

    @classmethod
    def get_active_users(cls) -> List[T]:
        return cls.filter(is_active=True)

    @classmethod
    def activate_user(cls, user: User) -> User:
        return cls.update(user, is_active=True)

    @classmethod
    def deactivate_user(cls, user: User) -> User:
        return cls.update(user, is_active=False)

在视图中使用UserService：

from django.http import JsonResponse
from services.user_service import UserService

def get_users_by_age(request, age: int):
    users = UserService.get_users_by_age(age)
    user_list = []
    for user in users:
        user_list.append({"id": user.id, "username": user.username, "age": user.age})
    return JsonResponse({"users": user_list})

def get_active_users(request):
    users = UserService.get_active_users()
    user_list = []
    for user in users:
        user_list.append({"id": user.id, "username": user.username})
    return JsonResponse({"users": user_list})

def activate_user(request, user_id: int):
    user = UserService.get(pk=user_id)
    if user:
        UserService.activate_user(user)
        return JsonResponse({"status": "success", "message": "User activated"})
    else:
        return JsonResponse({"status": "error", "message": "User not found"})

def deactivate_user(request, user_id: int):
    user = UserService.get(pk=user_id)
    if user:
        UserService.deactivate_user(user)
        return JsonResponse({"status": "success", "message": "User deactivated"})
    else:
        return JsonResponse({"status": "error", "message": "User not found"})

现在，我们已经优化了基类和实战示例。基类中的方法更加健壮，使用@classmethod装饰器使得子类能够更好地继承和重写基类方法。同时，我们还使用了类型提示，以提高代码的可读性。在UserService类中，我们添加了一些与User相关的业务逻辑方法，如获取特定年龄的用户、获取激活用户、激活用户和停用用户等。这些方法可以根据实际项目需求进行扩展。

添加分页优化

请结合分页再对上述基类做优化

为了结合分页功能，我们可以在BaseService中添加一个paginate方法。这个方法将使用Django的分页类Paginator。以下是修改后的base_service.py和user_service.py：

  1. 修改base_service.py：
from typing import Type, TypeVar, List, Optional, Tuple
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import Model, QuerySet
from django.core.paginator import Paginator, Page

T = TypeVar("T", bound=Model)

class BaseService:
    model: Type[T]

    # ... 其他方法

    @classmethod
    def paginate(cls, queryset: QuerySet[T], page: int, items_per_page: int) -> Tuple[Page[T], int]:
        paginator = Paginator(queryset, items_per_page)
        try:
            page_obj = paginator.page(page)
        except Exception:
            page_obj = paginator.page(1)
            page = 1

        return page_obj, page

    # 可以根据需要添加其他通用方法
  1. 修改user_service.py：
from typing import List
from .base_service import BaseService, T
from your_app.models import User  # 从你的应用中导入User模型

class UserService(BaseService):
    model = User

    # ... 其他方法

    @classmethod
    def get_users_by_age_paginated(cls, age: int, page: int, items_per_page: int) -> Tuple[List[T], int]:
        queryset = cls.filter(age=age)
        return cls.paginate(queryset, page, items_per_page)
  1. 在视图中使用UserService分页功能：
from django.http import JsonResponse
from services.user_service import UserService

def get_users_by_age_paginated(request, age: int):
    page = request.GET.get("page", 1)
    items_per_page = request.GET.get("items_per_page", 10)
    users, current_page = UserService.get_users_by_age_paginated(age, int(page), int(items_per_page))
    user_list = []
    for user in users:
        user_list.append({"id": user.id, "username": user.username, "age": user.age})
    return JsonResponse({"users": user_list, "page": current_page})

现在，在BaseService中添加了一个paginate方法，它接受一个查询集、当前页数和每页的项目数，并返回一个分页后的项目列表。在UserService中，我们添加了一个get_users_by_age_paginated方法，它使用基类中的paginate方法实现分页。最后，在视图中，我们可以直接调用UserService.get_users_by_age_paginated方法获取特定年龄的用户列表并分页。

上述分页会先查出所有数据，然后再做分页，如果数据量大查询会很慢，请对上述分页做优化

实际上，Django 的 Paginator 类在处理大量数据时是非常高效的。Paginator 不会一次性查询所有数据，而是按需查询数据。当你请求一个特定页的数据时，它会根据该页的索引和每页的项数来查询相应的数据。因此，它可以在不消耗大量内存的情况下处理大量数据。以下是一个更详细的例子，说明Paginator是如何工作的：

首先，我们创建一个包含10万条数据的示例数据表：

from django.db import models

class LargeData(models.Model):
    field1 = models.CharField(max_length=100)
    field2 = models.CharField(max_length=100)

然后我们创建一个LargeDataService类，用于处理LargeData模型：

from .base_service import BaseService, T
from your_app.models import LargeData

class LargeDataService(BaseService):
    model = LargeData

    @classmethod
    def get_data_paginated(cls, page: int, items_per_page: int) -> Tuple[List[T], int]:
        queryset = cls.filter()
        return cls.paginate(queryset, page, items_per_page)

在视图中使用LargeDataService分页功能：

from django.http import JsonResponse
from services.large_data_service import LargeDataService

def get_large_data_paginated(request):
    page = request.GET.get("page", 1)
    items_per_page = request.GET.get("items_per_page", 10)
    data, current_page = LargeDataService.get_data_paginated(int(page), int(items_per_page))
    data_list = [{"id": item.id, "field1": item.field1, "field2": item.field2} for item in data]
    return JsonResponse({"data": data_list, "page": current_page})

在这个示例中，我们有一个包含10万条数据的表。当我们请求特定页面时，Paginator 只会查询所需的数据，而不是一次性查询所有数据。因此，即使数据量很大，查询速度也很快。

如果您仍然担心性能问题，可以考虑使用第三方分页库，例如 django-pagination 或 django-el-pagination。这些库提供了更多的分页选项和性能优化。

完整封装

以下是根据谷歌编码规范要求的基类代码及注释说明：
base_service.py

from typing import Type, TypeVar, List, Optional, Tuple
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import Model, QuerySet
from django.core.paginator import Paginator, Page

T = TypeVar("T", bound=Model)

class BaseService:
    """
    BaseService是一个基类，提供通用的数据库访问方法。
    它用于定义与数据访问相关的通用方法，供子类继承和使用。
    """
    model: Type[T]

    @classmethod
    def get(cls, *args, **kwargs) -> Optional[T]:
        """
        通过给定的参数查找并返回一个模型实例，如果找不到则返回None。

        Args:
            *args: 传递给objects.get()方法的位置参数。
            **kwargs: 传递给objects.get()方法的关键字参数。

        Returns:
            如果找到，返回找到的模型实例；否则返回None。
        """
        try:
            instance = cls.model.objects.get(*args, **kwargs)
        except ObjectDoesNotExist:
            instance = None
        return instance

    @classmethod
    def filter(cls, *args, **kwargs) -> QuerySet[T]:
        """
        通过给定的参数查找并返回一个查询集。

        Args:
            *args: 传递给objects.filter()方法的位置参数。
            **kwargs: 传递给objects.filter()方法的关键字参数。

        Returns:
            返回一个查询集，包含满足条件的模型实例。
        """
        return cls.model.objects.filter(*args, **kwargs)

    @classmethod
    def create(cls, **kwargs) -> T:
        """
        使用给定的参数创建并返回一个新的模型实例。

        Args:
            **kwargs: 传递给objects.create()方法的关键字参数。

        Returns:
            返回创建的新模型实例。
        """
        return cls.model.objects.create(**kwargs)

    @classmethod
    def update(cls, instance: T, **kwargs) -> T:
        """
        使用给定的参数更新一个模型实例，并将其保存到数据库。

        Args:
            instance: 要更新的模型实例。
            **kwargs: 包含要更新的属性和值的字典。

        Returns:
            返回更新后的模型实例。
        """
        for attr, value in kwargs.items():
            setattr(instance, attr, value)
        instance.save()
        return instance

    @classmethod
    def delete(cls, instance: T) -> None:
        """
        删除一个模型实例。

        Args:
            instance: 要删除的模型实例。
        """
        instance.delete()

    @classmethod
    def paginate(cls, queryset: QuerySet[T], page: int, items_per_page: int) -> Tuple[Page[T], int]:
        """
        对给定的查询集进行分页。

        Args:
            queryset: 要分页的查询集。
            page: 要请求的页数。
            items_per_page: 每页显示的项数。

        Returns:
            返回一个元组，包含一个分页后的项目列表和实际的当前页数。
        """
        paginator = Paginator(queryset, items_per_page)
        try:
            page_obj = paginator.page(page)
        except Exception:
            page_obj = paginator.page(1)
            page = 1

        return page_obj, page

    # 可以根据需要添加其他通用方法

这是一个按照谷歌编码规范要求编写的BaseService基类。在这个基类中，我们为每个方法添加了详细的文档字符串，说明了方法的作用、输入参数和返回值。这样一来，其他开发者在使用这个基类时可以更容易地理解每个方法的作用和用法。

使用示例

请根据上述基类，给出每个方法使用的实战示例
假设我们有一个Blog模型，用于表示一个简单的博客。我们将创建一个BlogService类来使用BaseService中的方法。以下是Blog模型和BlogService类的示例：

  1. models.py：
from django.db import models

class Blog(models.Model):
    title = models.CharField(max_length=100)
    content = models.TextField()
    is_published = models.BooleanField(default=False)
  1. blog_service.py：
from .base_service import BaseService, T
from your_app.models import Blog

class BlogService(BaseService):
    model = Blog

现在我们可以在视图中使用BlogService来实现各种业务逻辑。以下是一些实战示例：

  1. 创建一个新的博客：
from services.blog_service import BlogService

def create_blog(request):
    title = request.POST.get("title")
    content = request.POST.get("content")

    blog = BlogService.create(title=title, content=content)
    # blog现在是一个新创建的Blog实例
  1. 根据ID获取一个博客：
from services.blog_service import BlogService

def get_blog(request, blog_id):
    blog = BlogService.get(pk=blog_id)
    # blog是找到的Blog实例，如果没有找到则为None
  1. 更新一个博客的标题和内容：
from services.blog_service import BlogService

def update_blog(request, blog_id):
    title = request.POST.get("title")
    content = request.POST.get("content")

    blog = BlogService.get(pk=blog_id)
    if blog:
        updated_blog = BlogService.update(blog, title=title, content=content)
        # updated_blog现在是更新后的Blog实例
    else:
        # blog不存在的处理逻辑
  1. 删除一个博客：
from services.blog_service import BlogService

def delete_blog(request, blog_id):
    blog = BlogService.get(pk=blog_id)
    if blog:
        BlogService.delete(blog)
        # blog已被删除
    else:
        # blog不存在的处理逻辑
  1. 获取所有已发布的博客：
from services.blog_service import BlogService

def get_published_blogs(request):
    published_blogs = BlogService.filter(is_published=True)
    # published_blogs是一个包含所有已发布博客的查询集
  1. 分页获取博客：
from services.blog_service import BlogService

def get_blogs_paginated(request):
    page = request.GET.get("page", 1)
    items_per_page = request.GET.get("items_per_page", 10)

    blogs, current_page = BlogService.paginate(BlogService.filter(), int(page), int(items_per_page))
    # blogs是一个包含分页后的博客的列表

这些实战示例展示了如何在视图中使用BlogService来实现各种业务逻辑。BaseService中的通用方法可以方便地应用于不同的业务场景。

代码释疑

请解释该代码：T = TypeVar("T", bound=Model)

这行代码定义了一个类型变量（Type Variable）T，它是Python类型提示（type hints）中的一个概念。类型变量通常用于泛型编程，表示一个可变的类型。在这个例子中，类型变量T是用于BaseService类的泛型参数。

TypeVar是Python标准库typing模块中的一个函数，用于创建类型变量。它接受两个参数：

  1. 类型变量的名字，这里是"T"。通常，类型变量的名字是一个大写字母，如T、U、V等。
  2. bound参数，表示类型变量的上界。在这个例子中，bound=Model意味着T的类型必须是Model类或其子类。这个约束确保了BaseService只能用于处理Model类及其子类的实例。

在BaseService类中，我们用类型变量T作为泛型参数来表示我们处理的模型类型。这使得BaseService能够处理任何继承自Model的子类，并且类型提示将正确地传递给子类。当我们为特定的模型创建服务类时，例如UserService或BlogService，类型变量T将被实际模型类替换，如User或Blog。这样一来，在服务类中，我们就可以使用正确的模型类型进行类型提示，使得代码更加清晰和健壮。

请解释： model: Type[T]
model: Type[T]是一个类变量注解，表示BaseService类中的model变量的类型。在这里，Type[T]意味着model变量的类型应该是一个T类型的类。T是一个类型变量，如前面所解释的，它的上界是Model。因此，T代表了一个继承自Model的类。

在BaseService的子类中，model变量将被设置为一个具体的Model子类，例如User、Blog等。这样一来，在BaseService的子类中，我们就可以使用model变量来访问模型的静态方法，如objects.get()、objects.filter()等。

例如，在一个名为BlogService的子类中，我们可以这样设置model变量：

from your_app.models import Blog

class BlogService(BaseService):
    model = Blog

在这个例子中，BlogService继承自BaseService，并将model变量设置为Blog类。这意味着在BlogService中，我们可以使用BaseService提供的通用方法来操作Blog模型。由于T类型变量的约束，我们可以确保model变量只能被设置为Model类及其子类。这样可以避免错误地设置model变量为其他类型。

在多线程中，上述基类的业务实现service会出现混用吗

在多线程环境下，上述基类BaseService的业务实现应该是线程安全的。原因如下：

  1. 所有的方法都是类方法（classmethod），这意味着这些方法是绑定到类上的，而不是实例。因此，在多线程环境下，每个线程都将使用相同的方法。
  2. 方法中没有使用全局变量或者类变量来存储状态，而是将状态作为方法参数传递。这可以避免多个线程之间共享可变状态，从而导致数据混乱。
  3. Django 的查询集（QuerySet）是惰性求值的，它们在被实际需要时才会计算结果。在多线程环境下，每个线程都可以有自己独立的查询集，而不会互相干扰。
  4. 在上述基类中，我们没有直接修改数据库记录。所有的数据库操作都是通过 Django 的 ORM 进行的，该 ORM 会在必要时处理并发问题。

尽管上述基类应该是线程安全的，但在实际应用中，处理多线程问题时还是要小心。例如，在使用基类的方法时，尽量避免在多个线程中共享可变状态。另外，根据实际业务需求，可能还需要对某些操作进行加锁以确保数据一致性。

总之，在多线程环境下使用上述基类不会出现混用问题，但实际业务逻辑实现时仍需注意线程安全性。

为者常成，行者常至

自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）