从0到1开发自动化运维平台接口文档分页视图和权限配置
安装依赖pip install djangorestframework-simplejwt pip install django-filter pip install coreapi pip install drf-yasg配置swagger接口文档
1、添加drf_yasg到settings.pyINSTALLED_APPS = [ ... "rest_framework", "drf_yasg", "cmdb.apps.CmdbConfig", ]
2、配置路由from rest_framework import permissions from drf_yasg.views import get_schema_view from drf_yasg import openapi schema_view = get_schema_view( openapi.Info( title="DevOps运维平台", default_version="v1", description="DevOps运维平台 接口文档", terms_of_service="", contact=openapi.Contact(email="qqing_lai@hotmail.com"), license=openapi.License(name="BSD License"), ), public=True, permission_classes=(permissions.AllowAny,), ) ... urlpatterns = [ path("apidoc/", schema_view.with_ui("swagger", ... ]restframework配置# drf配置 REST_FRAMEWORK = { # 自定义分页 "DEFAULT_PAGINATION_CLASS": "common.extends.pagination.CustomPagination", "PAGE_SIZE": 20, # 用户登陆认证方式 "DEFAULT_AUTHENTICATION_CLASSES": ( "rest_framework_simplejwt.authentication.JWTAuthentication", "rest_framework.authentication.SessionAuthentication", "rest_framework.authentication.BasicAuthentication", ), # 全局权限拦截 "DEFAULT_PERMISSION_CLASSES": ( "common.extends.permissions.RbacPermission", ), "DEFAULT_SCHEMA_CLASS": "rest_framework.schemas.coreapi.AutoSchema", "DEFAULT_FILTER_BACKENDS": ( "django_filters.rest_framework.DjangoFilterBackend", "rest_framework.filters.SearchFilter", "rest_framework.filters.OrderingFilter", ), "DEFAULT_RENDERER_CLASSES": ["rest_framework.renderers.JSONRenderer", "rest_framework.renderers.BrowsableAPIRenderer"] if DEBUG else [ "rest_framework.rend分页扩展#!/usr/bin/env python # -*- encoding: utf-8 -*- """ @author : Charles Lai @file : pagination.py @time : 2023/03/26 15:22 @contact : qqing_lai@hotmail.com """ # here put the import lib from rest_framework.pagination import PageNumberPagination, LimitOffsetPagination from rest_framework.response import Response from rest_framework import status class CustomPagination(PageNumberPagination): def get_paginated_response(self, data): return Response({"data": {"list": data, "total": self.page.paginator.count, "next": self.get_next_link(), "previous": self.get_previous_link()}, "success": True, "errorCode": 0, "errorMessage": None}, status=status.HTTP_200_OK) 自定义公共视图#!/usr/bin/env python # -*- encoding: utf-8 -*- """ @author : Charles Lai @file : viewsets.py @time : 2023/03/26 14:42 @contact : qqing_lai@hotmail.com """ # here put the import lib import inspect from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework import status from rest_framework import viewsets from rest_framework import pagination from rest_framework.settings import api_settings from rest_framework.filters import OrderingFilter from django.db.models.query import QuerySet from django.db.models import ProtectedError from django.core.cache import cache import pytz import logging logger = logging.getLogger(__name__) def ops_response(data, success=True, errorCode=0, errorMessage=None, status=status.HTTP_200_OK): """ 返回自定义 data列表数据格式:{ list: [ ], current?: number, pageSize?: number, total?: number, } """ return Response({"data": data, "success": success, "errorCode": errorCode, "errorMessage": errorMessage}, status=status) class AutoModelViewSet(viewsets.ModelViewSet): """ A viewset that provides default `create()`, `retrieve()`, `update()`, `partial_update()`, `destroy()` and `list()` actions. """ permission_classes = [IsAuthenticated] permission_classes_by_action = {} filter_backends = (OrderingFilter, ) def __init__(self, *args, **kwargs): if not hasattr(self, "queryset"): raise AttributeError("必须定义 类属性 queryset") if not hasattr(self, "serializer_class"): raise AttributeError("必须定义 类属性 serializer_class") super().__init__(*args, **kwargs) def get_serializer(self, *args, **kwargs): """ 重写 get_serializer 类,用来支持自动获取不同的 serializer_class 例子: list 方法, 设置一个serializer_list_class, 则调用get_serializer的时候, 优先获取 命名格式 serializer_{call_func_name}_class :param args: :param kwargs: :return: """ call_func_name = inspect.stack()[1][3] serializer_class = getattr(self, f"serializer_{call_func_name}_class", None) if not serializer_class: serializer_class = self.get_serializer_class() kwargs["context"] = self.get_serializer_context() return serializer_class(*args, **kwargs) def get_object(self): return super(AutoModelViewSet, self).get_object() def get_permissions(self): try: return [permission() for permission in self.permission_classes_by_action[self.action]] except KeyError: return [permission() for permission in self.permission_classes] def get_permission_from_role(self, request): try: perms = request.user.roles.values( "permissions__method", ).distinct() return [p["permissions__method"] for p in perms] except (AttributeError, TypeError): return [] def extend_filter(self, queryset): return queryset def get_queryset(self): assert self.queryset is not None, ( ""%s" should either include a `queryset` attribute, " "or override the `get_queryset()` method." % self.__class__.__name__ ) queryset = self.extend_filter(self.queryset) if isinstance(queryset, QuerySet): queryset = queryset.all() return queryset.distinct() def create(self, request, *args, **kwargs): try: request.data["name"] = request.data["name"].strip(" ").replace(" ", "-") except BaseException as e: logger.debug("exception ", str(e)) serializer = self.get_serializer(data=request.data) if not serializer.is_valid(): return ops_response({}, success=False, errorCode=40000, errorMessage=serializer.errors) try: self.perform_create(serializer) except BaseException as e: return ops_response({}, success=False, errorCode=50000, errorMessage=str(e), status=status.HTTP_500_INTERNAL_SERVER_ERROR) return ops_response(serializer.data) def list(self, request, pk=None, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) page_size = request.query_params.get("page_size", None) if not page_size: page_size = api_settings.PAGE_SIZE pagination.PageNumberPagination.page_size = page_size page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return ops_response({"list": serializer.data, "total": queryset.count()}) def update(self, request, *args, **kwargs): instance = self.get_object() partial = kwargs.pop("partial", False) try: request.data["name"] = request.data["name"].strip(" ").replace(" ", "-") except BaseException as e: logger.warning(f"不包含name字段: {str(e)}") serializer = self.get_serializer(instance, data=request.data, partial=partial) if not serializer.is_valid(): return ops_response({}, success=False, errorCode=40000, errorMessage=serializer.errors) try: self.perform_update(serializer) except BaseException as e: return ops_response({}, success=False, errorCode=50000, errorMessage=str(e)) if getattr(instance, "_prefetched_objects_cache", None): instance._prefetched_objects_cache = {} data = {"data": serializer.data, "status": "success", "code": 20000} return ops_response(serializer.data) def retrieve(self, request, *args, **kwargs): instance = self.get_object() serializer = self.get_serializer(instance) return ops_response(serializer.data) def destroy(self, request, *args, **kwargs): instance = self.get_object() try: self.perform_destroy(instance) except ProtectedError: # 存在关联数据,不可删除 return ops_response({}, success=False, errorCode=40300, errorMessage="存在关联数据,禁止删除!") except BaseException as e: logger.exception(f"删除数据发生错误 {e}, {e.__class__}") return ops_response({}, success=False, errorCode=50000, errorMessage=f"删除异常: {str(e)}") return ops_response("删除成功") class AutoModelParentViewSet(AutoModelViewSet): def get_queryset(self): assert self.queryset is not None, ( ""%s" should either include a `queryset` attribute, " "or override the `get_queryset()` method." % self.__class__.__name__ ) queryset = self.extend_filter(self.queryset) if self.action == "list": if not self.request.query_params.get("search"): queryset = queryset.filter(parent__isnull=True) if isinstance(queryset, QuerySet): queryset = queryset.all() return queryset.distinct() 自定义权限校验#!/usr/bin/env python # -*- encoding: utf-8 -*- """ @author : Charles Lai @file : permissions.py @time : 2023/03/26 15:27 @contact : qqing_lai@hotmail.com """ # here put the import lib from rest_framework.permissions import BasePermission from config import platform import logging logger = logging.getLogger(__name__) class RbacPermission(BasePermission): """ 自定义权限 """ @classmethod def check_is_admin(cls, request): return request.user.is_authenticated and request.user.roles.filter(name="管理员").count() > 0 @classmethod def get_permission_from_role(cls, request): try: perms = request.user.roles.values( "permissions__method", ).distinct() return [p["permissions__method"] for p in perms] except AttributeError: return [] def _has_permission(self, request, view): """ 权限获取方式 从 perms_map 中获取, 通过 request.method, http 请求方法来获取对应权限点 1. 默认格式 perms_map = ( {"*": ("admin", "管理员")}, {"*": ("k8s_all", "k8s管理")}, {"get": ("k8s_list", "查看k8s")}, {"post": ("k8s_create", "创建k8s")}, {"put": ("k8s_edit", "编辑k8s")}, {"delete": ("k8s_delete", "删除k8s")} ) 2. 自定义方法格式 perms_map = ( {"get_test_data": ("get_test_data", "获取测试数据")}, ) 此时 格式为 {http请求方法}_{ViewSet自定义action} :param request: rest_framework request 对象 :param view: rest_framework view 对象 :return: """ _method = request._request.method.lower() url_whitelist = platform["whitelist"] if platform else [] path_info = request.path_info for item in url_whitelist: url = item["url"] if url in path_info: logger.debug(f"请求地址 {path_info} 命中白名单 {url}, 放行") return True is_superuser = request.user.is_superuser # 超级管理员 或者 白名单模式 直接放行 if is_superuser: logger.debug(f"用户 {request.user} 是超级管理员, 放行 is_superuser = {is_superuser}") return True is_admin = RbacPermission.check_is_admin(request) perms = self.get_permission_from_role(request) # 不是管理员 且 权限列表为空的情况下, 直接拒绝 if not is_admin and not perms: logger.debug(f"用户 {request.user} 不是管理员 且 权限列表为空, 直接拒绝") return False perms_map = view.perms_map # 未配置权限映射的视图一律禁止访问 if not hasattr(view, "perms_map"): logger.debug(f"未配置权限映射的视图一律禁止访问 {view}") return False # _custom_method = None # default_funcs = ["create", "list", "retrieve", "update", "destroy"] action = view.action _custom_method = f"{_method}_{action}" for i in perms_map: logger.debug(f"perms_map item === {i}") for method, alias in i.items(): # 如果是管理员, 判断当前perms_map是否带有 {"*": ("admin", "管理员")} 标记,如果有, 则当前 ViewSet 所有方法全放行 if is_admin and (method == "*" and alias[0] == "admin"): logger.debug("管理员判断通过, 放行") return True # 如果带有某个模块的管理权限, 则当前模块所有方法都放行 if method == "*" and alias[0] in perms: logger.debug("模块管理权限 判断通过, 放行") return True # 判断自定义action的情况 # {"get_test_data": ("get_test_data", "获取测试数据")}, # {"*_test_data": ("get_test_data", "获取测试数据")}, if _custom_method and alias[0] in perms and (_custom_method == method or method == f"*_{action}"): logger.debug("自定义action权限 判断通过, 放行") return True # 判断是否拥有ViewSet 某个方法的权限, 有则放行 # {"get": ("workflow_list", "查看工单")}, if _method == method and alias[0] in perms: logger.debug(f"{method}方法权限 判断通过, 放行") return True logger.debug(f"{path_info} 没有符合条件的, 则默认禁止访问") return False def has_permission(self, request, view): res = self._has_permission(request, view) # 记录权限异常的操作 if not res: pass return res class AdminPermission(BasePermission): def has_permission(self, request, view): if RbacPermission.check_is_admin(request): return True return False class ObjPermission(BasePermission): """ 密码管理对象级权限控制 """ def has_object_permission(self, request, view, obj): perms = RbacPermission.get_permission_from_role(request) if "admin" in perms: return True elif request.user.id == obj.uid_id: return True 更新cmdb模块视图
我们之前已经完成的view_cmdb.py文件里,有几处需要更新:
1、将viewsets.ModelViewSet更改成公共视图里的AutoModelViewSet
2、将Response更改成ops_reponse,确保返回内容格式一致.运行项目(venv) ydevops-backend cd /home/charles/ydevops-backend ; /usr/bin/env /home/charles/ydevops-backend/venv/bin/python /home/charles/.vscode-server/extensions/ms-python.python-2023.5.10791008/pythonFiles/lib/python/debugpy/adapter/../../debugpy/launcher 41223 -- /home/charles/ydevops-backend/manage.py runserver 0.0.0.0:9000 Watching for file changes with StatReloader Performing system checks... System check identified no issues (0 silenced). March 26, 2023 - 18:50:30 Django version 4.1.7, using settings "devops_backend.settings" Starting development server at http://0.0.0.0:9000/ Quit the server with CONTROL-C.
访问接口文档 http://localhost:9000/apidoc/
在这个页面,我们可以做一些CRUD操作,如查看环境(由于数据量小,可以先把settings.py里的默认PAGE_SIZE改为1)
Okay,今天先到这吧...
2021年哪些事件让你印象深刻盘点2021年十大电竞新闻2021年是电子竞技深刻融入主流文化的一年。电竞入亚意味着电子竞技正式进入奥林匹克赛事大家庭,EDG夺冠则成为本年度最为出圈的电竞新闻。以下是我们总结的过去一年中,对电竞影响最大的
王者荣耀发布S26赛季更新时间王者荣耀S25赛季即将结束啦,王者荣耀官方在昨日发布了2022千色云中赛年S26赛季玉城之子更新时间为1月6日正式上线,截止目前的官方爆料中可以得知,S26赛季有以下几方面的调整。
LOL新版本改动预告船长皎月加强,琴女星蚀等多件装备削弱今天早上拳头的设计师在社交媒体上公布了LOL新版本的改动预告,具体分为三个部分。首先是加强船长,皎月,自然之力。然后是削弱琴女,雷克塞,星蚀,智慧末刃,不朽盾弓。最后是系统方面的改
游戏王禁卡目录未来融合接触过OTCG的牌佬应该都知道,在游戏王的历史上出现过一个神仙打架的超主流时期,那就是征龙与魔导大战。但是这张卡成为禁卡的时间线则在那次大战的前面一点点,现在回想起来或许在K社决定
LPL新赛季各选手评级图火了,UZI的评级真实,全神班BLG沦为笑话相信关注LPL赛事的小伙伴们都知道,S11赛季已经成为过去式,S12即将于1月10号正式打响。经过冬季转会期的LPL各战队将以全新的阵容征战。那么新赛季各战队的实力究竟如何呢,近日
PicoVR深度体验字节元宇宙氪金不足?作者丨风衔月维克多雨果曾说,今天的空想,就是明天的现实,元宇宙(Metaverse),正是人类对下一代互联网形态的设想。2021年,元宇宙引爆了整个互联网科技圈,成为了一个年度热词
一个游戏商人的退出哈喽,看我!曾经的一个游戏小商人。一边玩游戏一边在游戏中倒卖游戏物资既嗨皮又能赚钱感觉找到了成功之路这些都是以前交易的一些物资通俗的说就是个二道贩子,理想条件下是低买高卖,前期准备
网友自制LPL战队选手评级!EDG4人S级,TES双C是S级!Uzi唯一G级距离LPL春季赛开赛的时间越来越近了,最近除了外媒公布各种各样的排名之外,抗压背锅吧的老哥们也整理了一份榜单,给LPL各大战队的选手实力评级,最终得出了战队评级,一起来看看这份榜单
2。4版本申鹤值得抽吗?我直接得出结论,不值得!我是平民玩家,不是氪佬,申鹤这个角色零命是被严重下毒的角色,抽到零命也玩不开心,零申鹤是80能量的角色,在我看来,要求那么高能量的角色,抽到后,必须得上充能
剑网3官方终于想起承诺?1月上架外观是往年四分之一众所周知,剑网3这个游戏的外观在总收入中是占了非常大的比例的,这也就导致了近几年外观越出越多越出越频繁。而在今年的7月18日也因为玩家的激烈反抗正式宣布了降低外观发售频率和数量,即
尼泊尔的大富豪开直升飞机回老家,见人给一百,涨见识了尼泊尔村长的大哥是大富豪,村里人尽皆知。喜马拉雅山脉美丽的喜马拉雅山脉,群山环绕,远处的雪山如一道耀眼的星光照亮了整个山谷。浩瀚的云海,神秘的佛光,吸引了世界各地的游客前来观光。络