给请求封装通用参数
使用中间件
中间件可以在请求进入视图之前或响应返回给客户端之前执行代码。你可以创建一个自定义中间件来添加通用参数。
class CommonParamsMiddleware:def __init__(self, get_response):self.get_response = get_responsedef __call__(self, request):request.common_params = {'param1': 'value1','param2': 'value2',}response = self.get_response(request)return response
然后在你的settings.py中添加这个中间件：
MIDDLEWARE = [# ..'path.to.your.middleware.CommonParamsMiddleware',
]
在你的视图中，你可以通过request.common_params来访问这些参数。
使用自定义视图基类
如果你想要在视图层面添加通用参数，你可以创建一个自定义的视图基类，并在其他视图中继承它。
# views.py
from rest_framework.views import APIViewclass CommonParamsAPIView(APIView):def initialize_request(self, request, *args, **kwargs):request = super().initialize_request(request, *args, **kwargs)# 在这里添加你的通用参数request.common_params = {'param1': 'value1','param2': 'value2',}return request# 然后在你的其他视图中继承这个基类
class MyView(CommonParamsAPIView):def get(self, request, *args, **kwargs):# 使用通用参数common_params = request.common_params# ...
这样，任何继承CommonParamsAPIView的视图都会自动获得这些通用参数。
使用视图装饰器
如果你不想创建一个新的基类，你也可以使用装饰器来修改视图的行为。
# decorators.py
from functools import wrapsdef add_common_params(view_func):@wraps(view_func)def _decorated(view_instance, request, *args, **kwargs):# 在这里添加你的通用参数request.common_params = {'param1': 'value1','param2': 'value2',}return view_func(view_instance, request, *args, **kwargs)return _decorated# 在你的视图中使用装饰器
from rest_framework.decorators import api_view
from .decorators import add_common_params@api_view(['GET'])
@add_common_params
def my_view(request):# 使用通用参数common_params = request.common_params# ...
选择哪种方法取决于你的具体需求和偏好。中间件适用于全局的请求处理，自定义视图基类和装饰器则提供了更细粒度的控制。
打印所有内置属性/方法
def log_request(self, request):for attr in dir(request):if not attr.startswith('_'):# 检查属性是否存在，避免访问已经被移除的属性if hasattr(request, attr):try:value = getattr(request, attr, None)if not callable(value):print(f"{attr} : {value}")except Exception as e:print(f"Error accessing {attr}: {e}")
callable是一个内置函数，它用来检查一个对象是否是“可调用的”。一个对象如果是可调用的，意味着你可以像函数一样使用它，即可以在后面加上括号和参数来调用它。函数、方法、类以及实现了__call__方法的实例都是可调用的。
动态构建查找条件
def query_some(scope_type, scope_value):# 构建查询参数，其中 scope_type 是一个变量，它的值对应于模型中的字段名filter_kwargs = {scope_type: scope_value}objs = Project.objects.filter(**filter_kwargs).first()if not objs:return os.get("DEFAULT_RESULT")return objs.some_id
从yaml中读取真正的环境变量
# 必须参数
name: "MyNewSpace"
platform_url: os.getenv("XXX") #
app_code: os.getenv("YYY")# 非必需参数
desc: "Description of the new space."
def handle_env_var(value):# 尝试匹配 os.getenv("...") 格式并从环境变量中拿取真正的值if isinstance(value, str):match = re.match(r"os\.getenv\(['\"](.*?)['\"]\)", value)if match:return os.getenv(match.group(1))return valueelif isinstance(value, dict):for key, v in value.items():value[key] = handle_env_var(v)return valueelif isinstance(value, list):return [handle_env_var(v) for v in value]else:return value
动态构造序列化器内容：
class TASK_ID_KEY(Enum):TASK_ID = "task_tag"
class DemoQuerySerializer(serializers.Serializer):category = serializers.CharField(help_text="API对应的分类", max_length=255, default="test")# 动态添加字段def __init__(self, *args, **kwargs):super(DemoQuerySerializer, self).__init__(*args, **kwargs)self.fields[TASK_ID_KEY.TASK_ID.value] = serializers.CharField(help_text="task_tag_key", max_length=64)
异常处理
统一返回三段式处理函数
class ThirdPartResponse(Response):def __init__(self,result=True,message="",data=None,extra_data=None,status_code=status.HTTP_200_OK,code=None,):if message:message = "Error code: {}, {}".format(code, message)logger.error(message)standardized_data = {"result": result,"message": message,"data": data,}# 如果传入了extra_data，并且它是一个字典，则将其合并到输出字典中if extra_data and isinstance(extra_data, dict):standardized_data.update(extra_data)super().__init__(standardized_data, status_code)自定义异常类
from rest_framework import statusclass TokenException(Exception):status_code = status.HTTP_400_BAD_REQUESTcode = 1100error = ""def __init__(self, error):self.error = errorclass TokenAPIError(TokenException):code = 1101error = "get access_token from apigw failed"def __init__(self, error):super().__init__(error)class TokenNotExist(TokenException):code = 1102error = "access_token does not exist in db"def __init__(self, error):super().__init__(error)class APIException(Exception):status_code = status.HTTP_200_OKcode = 1200error = ""def __init__(self, error):self.error = errorclass ThirdPartException(APIException):code = 1201error = "第三方请求错误"def __init__(self, error):super().__init__(error)抛出和捕捉自定义异常
if not response["result"]: msg = "原样返回上游平台响应:{}".format(str(response)) logger.error(msg) raise ThirdPartyException(error=msg)
except IpHostFormatException as e: return ThreePartResponse(status_code=e.status_code, message=e.error, code=e.code)
捕捉系统异常
except Exception as e: msg = "reformat_global_vars error: {}".format(str(e)) logger.error(msg) return ThreePartResponse(message=msg)
正常返回三段式
return ThreePartResponse(data=serializer.data)
补充返回非三段式
return ThreePartResponse(data=data, extra_data=task_id_dict)