""" Data Policy Views - API views for data policy management. """ from rest_framework import viewsets, status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from rest_framework.views import APIView from django.apps import apps from .models import ( DataPolicy, DataScope, UserDataScope, RoleDataScope, DataAccessLog, ) from .serializers import ( DataPolicySerializer, DataPolicyCreateSerializer, DataScopeSerializer, UserDataScopeSerializer, RoleDataScopeSerializer, DataAccessLogSerializer, PolicyTestSerializer, ) from .services import DataPolicyService class DataPolicyViewSet(viewsets.ModelViewSet): """ API endpoint for managing data policies. Data policies define row-level security rules that filter what data users can see based on their context. """ permission_classes = [IsAuthenticated] def get_queryset(self): """Get data policies with optional filtering.""" queryset = DataPolicy.objects.all() # Filter by module module = self.request.query_params.get('module') if module: queryset = queryset.filter(module=module) # Filter by resource resource = self.request.query_params.get('resource') if resource: queryset = queryset.filter(resource=resource) # Filter by active status is_active = self.request.query_params.get('is_active') if is_active is not None: queryset = queryset.filter(is_active=is_active.lower() == 'true') return queryset.order_by('-priority', 'name') def get_serializer_class(self): """Get appropriate serializer.""" if self.action in ['create', 'update', 'partial_update']: return DataPolicyCreateSerializer return DataPolicySerializer @action(detail=False, methods=['get']) def policy_types(self, request): """Get available policy types.""" from .models import PolicyType types = [ {'value': choice[0], 'label': choice[1]} for choice in PolicyType.choices ] return Response(types) @action(detail=False, methods=['get']) def models_list(self, request): """Get list of models that can have policies.""" models = [] for app_config in apps.get_app_configs(): for model in app_config.get_models(): model_name = f"{model._meta.app_label}.{model._meta.model_name}" models.append({ 'name': model_name, 'verbose_name': str(model._meta.verbose_name), 'fields': [ f.name for f in model._meta.get_fields() if hasattr(f, 'name') ], }) return Response(sorted(models, key=lambda x: x['name'])) @action(detail=True, methods=['post']) def test(self, request, pk=None): """ Test a policy against sample data or existing records. Returns which records would be visible/hidden. """ policy = self.get_object() serializer = PolicyTestSerializer(data={ **request.data, 'policy_id': policy.id, 'model_name': f"{policy.module}.{policy.resource}", }) serializer.is_valid(raise_exception=True) try: model = apps.get_model(policy.module, policy.resource) sample_size = int(request.query_params.get('sample_size', 10)) original_qs = model.objects.all()[:sample_size] from apps.core.auth.models import User user = User.objects.get(id=serializer.validated_data['user_id']) filtered_qs = DataPolicyService.apply_policies( queryset=original_qs, user=user, module=policy.module, resource=policy.resource, ) return Response({ 'original_count': original_qs.count(), 'filtered_count': filtered_qs.count(), 'original_ids': list(original_qs.values_list('id', flat=True)), 'filtered_ids': list(filtered_qs.values_list('id', flat=True)), }) except Exception as e: return Response( {'error': str(e)}, status=status.HTTP_400_BAD_REQUEST ) @action(detail=True, methods=['post']) def duplicate(self, request, pk=None): """Duplicate a policy.""" policy = self.get_object() new_policy = DataPolicy.objects.create( tenant=policy.tenant, name=f"{policy.name} (Copy)", slug=f"{policy.slug}-copy", description=policy.description, module=policy.module, resource=policy.resource, policy_type=policy.policy_type, actions=policy.actions, conditions=policy.conditions, priority=policy.priority, is_allow=policy.is_allow, is_active=False, ) return Response( DataPolicySerializer(new_policy).data, status=status.HTTP_201_CREATED ) class DataScopeViewSet(viewsets.ModelViewSet): """ API endpoint for managing data scopes. """ serializer_class = DataScopeSerializer permission_classes = [IsAuthenticated] def get_queryset(self): queryset = DataScope.objects.all() scope_type = self.request.query_params.get('scope_type') if scope_type: queryset = queryset.filter(scope_type=scope_type) is_active = self.request.query_params.get('is_active') if is_active is not None: queryset = queryset.filter(is_active=is_active.lower() == 'true') return queryset.order_by('name') class UserDataScopeViewSet(viewsets.ModelViewSet): """ API endpoint for managing user data scope assignments. """ serializer_class = UserDataScopeSerializer permission_classes = [IsAuthenticated] def get_queryset(self): queryset = UserDataScope.objects.select_related('scope') user_id = self.request.query_params.get('user_id') if user_id: queryset = queryset.filter(user_id=user_id) return queryset.order_by('user', 'scope') class MyDataScopeView(APIView): """ API endpoint for getting current user's data scope. """ permission_classes = [IsAuthenticated] def get(self, request): """Get current user's effective data scope.""" scope = DataPolicyService.get_user_data_scope(request.user) return Response({ 'org_units': scope.get('org_units', []), 'has_global_access': scope.get('has_global_access', False), 'applied_policies': scope.get('applied_policies', []), }) class FieldVisibilityCheckView(APIView): """ API endpoint for checking field visibility. """ permission_classes = [IsAuthenticated] def get(self, request, model_name=None, field_name=None): """Check visibility of a specific field for current user.""" if not model_name or not field_name: return Response( {'error': 'model_name and field_name required'}, status=status.HTTP_400_BAD_REQUEST ) visibility = DataPolicyService.check_field_visibility( request.user, model_name, field_name ) return Response({ 'model': model_name, 'field': field_name, 'visibility': visibility, }) def post(self, request): """Check visibility of multiple fields.""" model_name = request.data.get('model') fields = request.data.get('fields', []) if not model_name: return Response( {'error': 'model required'}, status=status.HTTP_400_BAD_REQUEST ) result = {} for field in fields: result[field] = DataPolicyService.check_field_visibility( request.user, model_name, field ) return Response(result)