在DRF中实现一个可以读取request.data的自定义身份验证



我在我的模型上有一个外键,比如Patient和Doctor,它们指向一个Clinic类。所以,病人和医生应该只属于这个诊所。其他诊所不应该能够看到这些模型的任何细节。

模型是这样的:

class Clinic(models.Model):
clinicid = models.AutoField(primary_key=True, unique=True)
name = models.CharField(max_length=60, unique=True)
label = models.SlugField(max_length=25, unique=True)
email = models.EmailField(max_length=100, default='')
mobile = models.CharField(max_length=15, default='')
...
class Doctor(models.Model):
# Need autoincrement, unique and primary
docid = models.AutoField(primary_key=True, unique=True)
name = models.CharField(max_length=200)
username = models.CharField(max_length=15)
regid = models.CharField(max_length=15, default="", blank=True)
...
linkedclinic = models.ForeignKey(Clinic, on_delete=models.CASCADE)
class Patient(models.Model):
cstid = models.AutoField(primary_key=True, unique=True)
date_of_registration = models.DateField(default=timezone.now)
name = models.CharField(max_length=35, blank=False)
ageyrs = models.IntegerField(blank=True)
agemnths = models.IntegerField(blank=True)
dob = models.DateField(null=True, blank=True)
...
linkedclinic = models.ForeignKey(Clinic, on_delete=models.CASCADE)
class UserGroupMap(models.Model):
id = models.AutoField(primary_key=True, unique=True)
user = models.ForeignKey(
User, related_name='target_user', on_delete=models.CASCADE)
group = models.ForeignKey(UserGroup, on_delete=models.CASCADE)
clinic = models.ForeignKey(Clinic, on_delete=models.CASCADE)
...

从我的Vue应用中,我将使用Axios发布到使用DRF的django应用中,从而获得患者和医生的序列化数据。如果我尝试在函数视图中使用以下示例代码,则一切都可以正常工作:

@api_view(['GET', 'POST'])
def register_patient_vue(request):
if request.method == 'POST':
print("POST details", request.data)
data = request.data['registration_data']
serializer = customerSpecialSerializer(data=data)
if serializer.is_valid():
a = serializer.save()
print(serializer.data)
return Response(serializer.data, status=status.HTTP_201_CREATED)
else:
print("Serializer is notNot valid.")
print(serializer.errors)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

样本输出:

POST details {'registration_data': {'name': 'wczz', 'ageyrs': 21, 'agemonths': '', 'dob': '', 'gender': 'unspecified', 'mobile': '2', 'email': '', 'alternate': '', 'address': '', 'marital': 'unspecified', 'city': '', 'occupation': '', 'linkedclinic': 10}}
data: {'name': 'wczz', 'ageyrs': 21, 'agemonths': '', 'dob': '', 'gender': 'unspecified', 'mobile': '2', 'email': '', 'alternate': '', 'address': '', 'marital': 'unspecified', 'city': '', 'occupation': '', 'linkedclinic': 10}

但是,我需要通过特殊的自定义身份验证来验证请求。我有另一个名为UserGroupMap的类,它具有User和Clinic的外键,因此,如果映射中有诊所和用户的过滤器匹配,它将进行身份验证。否则,它将无法通过身份验证,并且不应该检索数据或保存序列化程序。

在我之前的简单的纯django项目中,我使用了一个自定义权限函数,并用它来装饰我的视图:

@handle_perm(has_permission_level, required_permission='EDIT_CLINICAL_RECORD', login_url='/clinic/')
def some_function(request, dept_id):
....
Some code which runs after authentication

它会使用以下语句:

def handle_perm(test_func, required_permission=None, login_url=None, redirect_field_name=REDIRECT_FIELD_NAME):
"""
Decorator for views that checks that the user passes the given test,
redirecting to the log-in page if necessary. The test should be a callable
that takes the user object and returns True if the user passes.
"""
def decorator(view_func):
@wraps(view_func)
def _wrapped_view(request, *args, **kwargs):
print(f"Required permission level is {required_permission}")
if has_permission_level(request, required_permission):
print("User has required permission level..Allowing entry.")
return view_func(request, *args, **kwargs)
print("FAILED! User does not have required permission level. Access blocked.")
path = request.build_absolute_uri()
resolved_login_url = resolve_url(login_url or settings.LOGIN_URL)
# If the login url is the same scheme and net location then just
# use the path as the "next" url.
login_scheme, login_netloc = urlparse(resolved_login_url)[:2]
current_scheme, current_netloc = urlparse(path)[:2]
if ((not login_scheme or login_scheme == current_scheme) and
(not login_netloc or login_netloc == current_netloc)):
path = request.get_full_path()
from django.contrib.auth.views import redirect_to_login
return redirect_to_login(
path, resolved_login_url, redirect_field_name)
return _wrapped_view
return decorator
def has_permission_level(request, required_permission, clinic=None):
print("has_permission_level was called.")
user = request.user
print(f'user is {user}')

clinic=clinic_from_request(request)
print(f"has_permission_level called with clinic:{clinic}")

if clinic is None:
print("clinic is none")
return HttpResponseRedirect('/accounts/login/')
group_maps = UserGroupMap.objects.filter(user=user, clinic=clinic)
print(f"No: of UserGroupMap memberships: {len(group_maps)}")
if len(group_maps) < 1:
# There are no UserGroupMap setup for the user. Kindly set them up.nHint:Admin>Manage users and groups>Users
return False
# Now checking Group memberships whether the user has any with permisison
for map in group_maps:
rolesmapped = GroupRoleMap.objects.filter(group=map.group)
if len(rolesmapped) < 1:
print(f"No permission roles.")
else:
for rolemap in rolesmapped:
print(f"{rolemap.role}", end=",")
if rolemap.role.name == required_permission:
print(
f"nAvailable role of [{map.group}] matched required permission of [{required_permission}] in {clinic.name} [Ok]")
return True
return False

我需要使用DRF构建一个自定义身份验证,以便它读取post数据,并检查linkedclinic值,并使用类似的逻辑。

我是这样开始的:

def has_permission_POST(request, required_permission, clinic=None):
print("has_permission_POST was called.")
user = request.user
print(f'user is {user}')
if request.method == 'POST':
print(request)
print(dir(request))
print("POST details: POST:", request.POST, "n")
print("POST details: data:", request.data, "n")
....
# Further logic to check the mapping
return True

else:
print("Not a valid POST")
return Response("INVALID POST", status=status.HTTP_400_BAD_REQUEST)
# And decorating my DRF view:
@handle_perm(has_permission_POST, required_permission='EDIT_CLINICAL_RECORD', login_url='/clinic/')
@api_view(['GET', 'POST'])
def register_patient_vue(request):
if request.method == 'POST':
print("POST details", request.data)
data = request.data['registration_data']

问题是,如果我运行这个,那么,has_permission_POST无法获得请求的值。数据,其中包含从我的前端发布的数据。我可以通过将@api_view(['GET', 'POST'])装饰器添加到has_permission_POST来解决这个问题。但是这会引入另一个错误,一个失败的断言:

AssertionError: Expected a `Response`, `HttpResponse` or `HttpStreamingResponse` to be returned from the view, but received a `<class 'bool'>`

一旦用@api_view装饰了has_permission_POST,就会发生这种情况。

So my problems:

  1. 如何为我的用例实现自定义身份验证?
  2. 如果我这样做,通过使用这个自定义的has_permission_level,我怎么能得到请求。在调用我的实际api视图之前,在这个函数中保存数据,以便我可以读取诊所id并执行我需要的权限检查。

我已经看了一下DRF提供的CustomAuthentication,但无法找到如何获得请求。自定义类中的数据参数。

感谢@MihaiChelaru,我能够找到一个解决我的问题的方法。

我通过扩展权限创建了自定义Permission类。BasePermission,并在特殊的has_permission函数中使用我的自定义逻辑。我更进一步,从请求中实现了Token的检查。一旦令牌通过身份验证,就可以从token表中的匹配令牌中获取用户。我发现在自定义权限类中,我可以读取Vue和Postman传递的完整request.data参数。一旦我读到这一点,我可以很容易地实现自定义检查用户权限,我的自定义模型。

class CustomerAccessPermission(permissions.BasePermission):
message = 'No permission to create new patient records'
def has_permission(self, request, view):
bearer_authorizn = request.META.get('HTTP_AUTHORIZATION')
try: #Different apps like POSTMAN, and Vue seem to use different strings while passing token
token = bearer_authorizn.split("Bearer ")[1]
except Exception as e:
try:
token = bearer_authorizn.split("Token ")[1]
except Exception as e:
raise NotAuthenticated('Did not get token in request')
try:
token_obj = Token.objects.get(key=token)
except self.model.DoesNotExist:
raise AuthenticationFailed('Invalid token')
if not token_obj.user.is_active:
raise AuthenticationFailed('User inactive or deleted')
print("Username is %s" % token_obj.user.username)
print("POST details", request.data)
linkedclinic_id = request.data['data']['linkedclinic']
clinic = Clinic.objects.get(clinicid=int(linkedclinic_id))
print("Clinic membership requested:", clinic)
group_maps = UserGroupMap.objects.filter(user=user, clinic=clinic)
print(f"No: of UserGroupMap memberships: {len(group_maps)}")
if len(group_maps) > 1:
return True
return False

@api_view(['POST'])
@permission_classes([CustomerAccessPermission])
def register_patient_vue(request):
logger.info('In register_patient_vue...')
...

最新更新