Skip to content

Instantly share code, notes, and snippets.

@IMBlues
Last active October 4, 2023 07:13
  • Star 2 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save IMBlues/e36e792159729f429f9abf656ba24d10 to your computer and use it in GitHub Desktop.
Make your Django REST framework supporting dependency injection
# -*- coding: utf-8 -*-
import functools
from collections import namedtuple
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Optional, Type
from django.conf import settings
from django.utils.module_loading import import_string
from rest_framework import status
try:
from drf_yasg.utils import swagger_auto_schema
SKIP_SWAGGER_SCHEMA = False
except ImportError:
SKIP_SWAGGER_SCHEMA = True
from rest_framework.serializers import BaseSerializer
if TYPE_CHECKING:
from rest_framework.request import Request
ResponseParams = namedtuple("ResponseParams", "data,params")
@dataclass
class SerializerInjector:
"""A injector for injecting serializer as dependency"""
in_cls: Type[BaseSerializer]
out_cls: Type[BaseSerializer]
config: dict = field(default_factory=dict)
in_raw_params: dict = field(default_factory=dict)
out_raw_params: dict = field(default_factory=dict)
_default_config_value = {"data_from": "query_params", "return_validated_data": True, "remain_request": True}
def __post_init__(self):
self.in_raw_params = self.in_raw_params or dict(raise_exception=True)
self.config = self.config or {}
self.out_raw_params = self.out_raw_params or {}
for extend_config_name, default_value in self._default_config_value.items():
setattr(
self,
f"{extend_config_name}",
self.config.get(extend_config_name, default_value),
)
try:
self.resp_cls = import_string(settings.SERIALIZER_INJECTOR_RESP_CLS)
except AttributeError:
self.resp_cls = import_string("rest_framework.response.Response")
def __str__(self):
return f"Injector<In:{self.in_cls.__class__.__name__}, Out:{self.out_cls.__class__.__name__}>"
def update_out_params(self, params: dict):
"""Update out params"""
self.out_raw_params.update(params)
def get_serializer_instance(self, request: "Request") -> "BaseSerializer":
"""Get in serializer instance"""
slz_obj = self.in_cls(data=getattr(request, self.data_from)) # type: ignore
slz_obj.is_valid(**self.in_raw_params)
return slz_obj
def get_validated_data(self, request: "Request") -> dict:
"""Get validated data via in_serializer"""
return self.get_serializer_instance(request).validated_data
def get_in_params(self, request: "Request") -> dict:
"""Get extra params before view logic"""
if self.return_validated_data: # type: ignore
return {"validated_data": self.get_validated_data(request)}
else:
return {"serializer_instance": self.get_serializer_instance(request)}
def get_response(self, data):
"""Get Response data"""
return self.resp_cls(data=self.out_cls(data, **self.out_raw_params).data)
def serializer_inject(
in_cls: Type[BaseSerializer] = None,
out_cls: Type[BaseSerializer] = None,
config: Optional[dict] = None,
in_params: Optional[dict] = None,
out_params: Optional[dict] = None,
swagger_params: Optional[dict] = None,
):
def decorator_serializer_inject(func):
injector = SerializerInjector(in_cls, out_cls, config, in_params, out_params)
if not SKIP_SWAGGER_SCHEMA:
default_params = {}
if in_cls:
if injector.data_from == "query_params":
default_params = {"query_serializer": in_cls()}
else:
default_params = {"request_body": in_cls()}
if out_cls:
default_params.update({"responses": {status.HTTP_200_OK: out_cls()}})
default_params.update(swagger_params or {})
func = swagger_auto_schema(**default_params)(func)
@functools.wraps(func)
def decorated(*args, **kwargs):
in_content = {}
if in_cls:
in_content.update(**injector.get_in_params(args[1]))
if not injector.remain_request:
args = args[0] + args[1:]
original_data = func(*args, **kwargs, **in_content)
if not out_cls:
return original_data
# support runtime serializer params, like "context"
if isinstance(original_data, ResponseParams):
injector.update_out_params(original_data.params)
original_data = original_data.data
return injector.get_response(original_data)
return decorated
return decorator_serializer_inject
@IMBlues
Copy link
Author

IMBlues commented May 24, 2021

Examples

from some_path.inject import serializer_inject, ResponseParams


class LoginLogViewSet(GenericViewSet):
    @serializer_inject(
        in_cls=InSerializer,
        out_cls=OutSerializer,
        swagger_params=dict(tags=["fun"]),
    )
    def list(self, request, validated_data: dict):
        # insert your logic
        ...
        return results

class LoginLog2ViewSet(GenericViewSet):
    @serializer_inject(
        in_cls=InSerializer,
        out_cls=OutSerializer,
        swagger_params=dict(tags=["fun"]),
    )
    def list(self, request, validated_data: dict):
        # insert your logic
        ...
        # when you need to pass some context to serializer, use RepsonseParams to wrap different parts
        return ResponseParams(results, {"context": {"some-context": {}}})


class LoginLog3ViewSet(GenericViewSet):
    @serializer_inject(
        in_cls=InSerializer,
        out_cls=OutSerializer,
        # get data from query_params as default, you can change it manually,
        # and also you can use serializer instance instead of validated_data
        config={"data_from": "data", "return_validated_data": False}
        swagger_params=dict(tags=["fun"]),
    )
    def list(self, request, serializer_instance: InSerializer):
        # insert your logic
        ...
        # when you need to pass some context to serializer, use RepsonseParams to wrap different parts
        return ResponseParams(results, {"context": {"some-context": {}}})

class LoginLog4ViewSet(GenericViewSet):
    @serializer_inject(
        in_cls=InSerializer,
        out_cls=OutSerializer,
        # if you don't need request, remove it by setting config
        config={"remain_request": False}
        swagger_params=dict(tags=["fun"]),
    )
    def list(self, validated_data: dict):
        # insert your logic
        ...
        return results

@IMBlues
Copy link
Author

IMBlues commented Dec 15, 2021

可以在 蓝鲸用户管理项目 查看最新的更新

@kasir-barati
Copy link

@IMBlues Thanks. But I am wondered why such a mature - IMO - framework does not have DI by default. I mean it is really cool and every body needs it. Besides it is really peculiar that we have not bootstrapping process for our DRF app. I guess we have a problem in DRF, Django.

@IMBlues
Copy link
Author

IMBlues commented Jul 5, 2022

@IMBlues Thanks. But I am wondered why such a mature - IMO - framework does not have DI by default. I mean it is really cool and every body needs it. Besides it is really peculiar that we have not bootstrapping process for our DRF app. I guess we have a problem in DRF, Django.

Haha, I wondered it too. Dependency injection is not a particularly trendy concept, and the DRF community should consider supporting such syntactic sugar natively. Maybe sometime I'll raise an issue with the community and see if they'll accept it.

And I had updated some code in our SDK project(the code in this gist is not updated), you can install the sdk directly(althought we only have readme in chinese version, but I believe the code is pretty and easy reading) or just copy it to your own project. Enjoy!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment