254 lines
9.0 KiB
Python
254 lines
9.0 KiB
Python
"""
|
|
Core API views.
|
|
"""
|
|
|
|
from rest_framework import viewsets, filters, status
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from django_filters.rest_framework import DjangoFilterBackend
|
|
from django.db.models import Q
|
|
from django.utils import timezone
|
|
from ..models import Tenant, AuditLogEntry, SystemConfiguration, SystemNotification, IntegrationLog
|
|
from .serializers import (
|
|
TenantSerializer, AuditLogEntrySerializer, SystemConfigurationSerializer,
|
|
SystemNotificationSerializer, IntegrationLogSerializer
|
|
)
|
|
|
|
|
|
class TenantViewSet(viewsets.ReadOnlyModelViewSet):
|
|
"""
|
|
Tenant API viewset.
|
|
"""
|
|
serializer_class = TenantSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
|
|
filterset_fields = ['organization_type', 'subscription_plan', 'is_active', 'is_trial']
|
|
search_fields = ['name', 'display_name', 'city', 'state']
|
|
ordering_fields = ['name', 'created_at']
|
|
ordering = ['name']
|
|
|
|
def get_queryset(self):
|
|
# Users can only see their own tenant
|
|
if hasattr(self.request, 'tenant') and self.request.tenant:
|
|
return Tenant.objects.filter(id=self.request.tenant.id)
|
|
return Tenant.objects.none()
|
|
|
|
@action(detail=True, methods=['get'])
|
|
def statistics(self, request, pk=None):
|
|
"""
|
|
Get tenant statistics.
|
|
"""
|
|
tenant = self.get_object()
|
|
stats = {
|
|
'user_count': tenant.get_user_count(),
|
|
'patient_count': tenant.get_patient_count(),
|
|
'is_trial_expired': tenant.is_trial_expired,
|
|
'subscription_plan': tenant.subscription_plan,
|
|
'max_users': tenant.max_users,
|
|
'max_patients': tenant.max_patients,
|
|
}
|
|
return Response(stats)
|
|
|
|
|
|
class AuditLogEntryViewSet(viewsets.ReadOnlyModelViewSet):
|
|
"""
|
|
Audit log entry API viewset.
|
|
"""
|
|
serializer_class = AuditLogEntrySerializer
|
|
permission_classes = [IsAuthenticated]
|
|
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
|
|
filterset_fields = [
|
|
'event_type', 'event_category', 'user', 'risk_level',
|
|
'hipaa_relevant', 'gdpr_relevant', 'is_successful'
|
|
]
|
|
search_fields = ['action', 'description', 'user_email', 'object_repr']
|
|
ordering_fields = ['timestamp', 'created_at']
|
|
ordering = ['-timestamp']
|
|
|
|
def get_queryset(self):
|
|
if hasattr(self.request, 'tenant') and self.request.tenant:
|
|
return AuditLogEntry.objects.filter(tenant=self.request.tenant)
|
|
return AuditLogEntry.objects.none()
|
|
|
|
@action(detail=False, methods=['get'])
|
|
def summary(self, request):
|
|
"""
|
|
Get audit log summary statistics.
|
|
"""
|
|
queryset = self.get_queryset()
|
|
now = timezone.now()
|
|
|
|
summary = {
|
|
'total_entries': queryset.count(),
|
|
'entries_today': queryset.filter(timestamp__date=now.date()).count(),
|
|
'entries_this_week': queryset.filter(
|
|
timestamp__gte=now - timezone.timedelta(days=7)
|
|
).count(),
|
|
'error_entries': queryset.filter(event_type='ERROR').count(),
|
|
'high_risk_entries': queryset.filter(risk_level='HIGH').count(),
|
|
'hipaa_relevant_entries': queryset.filter(hipaa_relevant=True).count(),
|
|
}
|
|
return Response(summary)
|
|
|
|
|
|
class SystemConfigurationViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
System configuration API viewset.
|
|
"""
|
|
serializer_class = SystemConfigurationSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
|
|
filterset_fields = ['category', 'data_type', 'is_active', 'is_readonly', 'is_sensitive']
|
|
search_fields = ['key', 'description']
|
|
ordering_fields = ['category', 'key', 'updated_at']
|
|
ordering = ['category', 'key']
|
|
|
|
def get_queryset(self):
|
|
if hasattr(self.request, 'tenant') and self.request.tenant:
|
|
return SystemConfiguration.objects.filter(
|
|
Q(tenant=self.request.tenant) | Q(tenant=None),
|
|
is_active=True
|
|
)
|
|
return SystemConfiguration.objects.none()
|
|
|
|
def perform_create(self, serializer):
|
|
serializer.save(
|
|
tenant=getattr(self.request, 'tenant', None),
|
|
updated_by=self.request.user
|
|
)
|
|
|
|
def perform_update(self, serializer):
|
|
serializer.save(updated_by=self.request.user)
|
|
|
|
@action(detail=False, methods=['get'])
|
|
def by_category(self, request):
|
|
"""
|
|
Get configurations grouped by category.
|
|
"""
|
|
queryset = self.get_queryset()
|
|
categories = {}
|
|
|
|
for config in queryset:
|
|
category = config.category
|
|
if category not in categories:
|
|
categories[category] = []
|
|
categories[category].append(self.get_serializer(config).data)
|
|
|
|
return Response(categories)
|
|
|
|
|
|
class SystemNotificationViewSet(viewsets.ModelViewSet):
|
|
"""
|
|
System notification API viewset.
|
|
"""
|
|
serializer_class = SystemNotificationSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
|
|
filterset_fields = [
|
|
'notification_type', 'priority', 'target_audience', 'is_active'
|
|
]
|
|
search_fields = ['title', 'message']
|
|
ordering_fields = ['priority', 'created_at']
|
|
ordering = ['-priority', '-created_at']
|
|
|
|
def get_queryset(self):
|
|
if hasattr(self.request, 'tenant') and self.request.tenant:
|
|
return SystemNotification.objects.filter(
|
|
Q(tenant=self.request.tenant) | Q(tenant=None)
|
|
)
|
|
return SystemNotification.objects.none()
|
|
|
|
def perform_create(self, serializer):
|
|
serializer.save(
|
|
tenant=getattr(self.request, 'tenant', None),
|
|
created_by=self.request.user
|
|
)
|
|
|
|
@action(detail=False, methods=['get'])
|
|
def active(self, request):
|
|
"""
|
|
Get active notifications.
|
|
"""
|
|
now = timezone.now()
|
|
queryset = self.get_queryset().filter(
|
|
is_active=True,
|
|
start_date__lte=now
|
|
).filter(
|
|
Q(end_date__gte=now) | Q(end_date=None)
|
|
)
|
|
|
|
serializer = self.get_serializer(queryset, many=True)
|
|
return Response(serializer.data)
|
|
|
|
@action(detail=True, methods=['post'])
|
|
def dismiss(self, request, pk=None):
|
|
"""
|
|
Dismiss a notification.
|
|
"""
|
|
notification = self.get_object()
|
|
if not notification.is_dismissible:
|
|
return Response(
|
|
{'error': 'This notification cannot be dismissed'},
|
|
status=status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
# Log the dismissal in audit log
|
|
from ..utils import AuditLogger
|
|
AuditLogger.log_event(
|
|
tenant=getattr(request, 'tenant', None),
|
|
event_type='UPDATE',
|
|
event_category='SYSTEM_ADMINISTRATION',
|
|
action='Dismiss Notification',
|
|
description=f'User dismissed notification: {notification.title}',
|
|
user=request.user,
|
|
content_object=notification,
|
|
request=request
|
|
)
|
|
|
|
return Response({'status': 'dismissed'})
|
|
|
|
|
|
class IntegrationLogViewSet(viewsets.ReadOnlyModelViewSet):
|
|
"""
|
|
Integration log API viewset.
|
|
"""
|
|
serializer_class = IntegrationLogSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
|
|
filterset_fields = [
|
|
'integration_type', 'direction', 'external_system', 'status'
|
|
]
|
|
search_fields = ['external_system', 'message_type', 'message_id']
|
|
ordering_fields = ['timestamp', 'created_at']
|
|
ordering = ['-timestamp']
|
|
|
|
def get_queryset(self):
|
|
if hasattr(self.request, 'tenant') and self.request.tenant:
|
|
return IntegrationLog.objects.filter(tenant=self.request.tenant)
|
|
return IntegrationLog.objects.none()
|
|
|
|
@action(detail=False, methods=['get'])
|
|
def summary(self, request):
|
|
"""
|
|
Get integration log summary statistics.
|
|
"""
|
|
queryset = self.get_queryset()
|
|
now = timezone.now()
|
|
|
|
summary = {
|
|
'total_logs': queryset.count(),
|
|
'logs_today': queryset.filter(timestamp__date=now.date()).count(),
|
|
'successful_logs': queryset.filter(status='SUCCESS').count(),
|
|
'failed_logs': queryset.filter(status='FAILED').count(),
|
|
'pending_logs': queryset.filter(status='PENDING').count(),
|
|
'integration_types': list(
|
|
queryset.values_list('integration_type', flat=True).distinct()
|
|
),
|
|
'external_systems': list(
|
|
queryset.values_list('external_system', flat=True).distinct()
|
|
),
|
|
}
|
|
return Response(summary)
|
|
|