402 lines
14 KiB
Python
402 lines
14 KiB
Python
"""
|
|
Common mixins for views across the Tenhal Multidisciplinary Healthcare Platform.
|
|
|
|
These mixins provide reusable functionality for tenant filtering, permissions,
|
|
audit logging, and HTMX support.
|
|
"""
|
|
|
|
from django.contrib import messages
|
|
from django.contrib.auth.mixins import UserPassesTestMixin
|
|
from django.core.exceptions import ImproperlyConfigured, PermissionDenied
|
|
from django.shortcuts import redirect, render
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from .models import AuditLog
|
|
|
|
|
|
class TenantFilterMixin:
|
|
"""
|
|
Mixin to automatically filter querysets by the current user's tenant.
|
|
|
|
Usage:
|
|
class MyListView(TenantFilterMixin, ListView):
|
|
model = MyModel
|
|
"""
|
|
|
|
def get_queryset(self):
|
|
"""Filter queryset by tenant."""
|
|
queryset = super().get_queryset()
|
|
|
|
# Check if model has tenant field
|
|
if hasattr(queryset.model, 'tenant'):
|
|
return queryset.filter(tenant=self.request.user.tenant)
|
|
|
|
return queryset
|
|
|
|
|
|
class RolePermissionMixin(UserPassesTestMixin):
|
|
"""
|
|
Mixin to check if user has required role(s) to access the view.
|
|
|
|
Usage:
|
|
class MyView(RolePermissionMixin, DetailView):
|
|
allowed_roles = ['ADMIN', 'DOCTOR']
|
|
"""
|
|
|
|
allowed_roles = [] # Override in subclass
|
|
|
|
def test_func(self):
|
|
"""Check if user's role is in allowed_roles."""
|
|
if not self.allowed_roles:
|
|
# If no roles specified, allow all authenticated users
|
|
return self.request.user.is_authenticated
|
|
|
|
return self.request.user.role in self.allowed_roles
|
|
|
|
def handle_no_permission(self):
|
|
"""Custom handling for permission denied."""
|
|
raise PermissionDenied(
|
|
f"You need one of these roles to access this page: {', '.join(self.allowed_roles)}"
|
|
)
|
|
|
|
|
|
class AuditLogMixin:
|
|
"""
|
|
Mixin to automatically log create, update, and delete actions.
|
|
|
|
Usage:
|
|
class MyCreateView(AuditLogMixin, CreateView):
|
|
model = MyModel
|
|
"""
|
|
|
|
def form_valid(self, form):
|
|
"""Log the action after successful form submission."""
|
|
response = super().form_valid(form)
|
|
|
|
# Determine action type
|
|
if hasattr(self, 'object') and self.object:
|
|
if hasattr(self.object, 'pk') and self.object.pk:
|
|
# Check if this is a new object or update
|
|
action = 'UPDATE' if form.instance.pk else 'CREATE'
|
|
else:
|
|
action = 'CREATE'
|
|
else:
|
|
action = 'UNKNOWN'
|
|
|
|
# Create audit log entry
|
|
try:
|
|
AuditLog.objects.create(
|
|
tenant=self.request.user.tenant,
|
|
user=self.request.user,
|
|
action=action,
|
|
model_name=self.model.__name__,
|
|
object_id=str(self.object.pk) if hasattr(self, 'object') else None,
|
|
changes=self._get_changed_fields(form),
|
|
ip_address=self._get_client_ip(),
|
|
user_agent=self.request.META.get('HTTP_USER_AGENT', '')[:255],
|
|
)
|
|
except Exception as e:
|
|
# Log error but don't fail the request
|
|
print(f"Failed to create audit log: {e}")
|
|
|
|
return response
|
|
|
|
def _get_changed_fields(self, form):
|
|
"""Get dictionary of changed fields."""
|
|
if not hasattr(form, 'changed_data'):
|
|
return {}
|
|
|
|
changes = {}
|
|
for field in form.changed_data:
|
|
if field in form.cleaned_data:
|
|
changes[field] = str(form.cleaned_data[field])
|
|
|
|
return changes
|
|
|
|
def _get_client_ip(self):
|
|
"""Get client IP address from request."""
|
|
x_forwarded_for = self.request.META.get('HTTP_X_FORWARDED_FOR')
|
|
if x_forwarded_for:
|
|
ip = x_forwarded_for.split(',')[0]
|
|
else:
|
|
ip = self.request.META.get('REMOTE_ADDR')
|
|
return ip
|
|
|
|
def delete(self, request, *args, **kwargs):
|
|
"""Log delete action."""
|
|
# Get object before deletion
|
|
self.object = self.get_object()
|
|
object_id = str(self.object.pk)
|
|
model_name = self.object.__class__.__name__
|
|
|
|
# Perform deletion
|
|
response = super().delete(request, *args, **kwargs)
|
|
|
|
# Create audit log entry
|
|
try:
|
|
AuditLog.objects.create(
|
|
tenant=request.user.tenant,
|
|
user=request.user,
|
|
action='DELETE',
|
|
model_name=model_name,
|
|
object_id=object_id,
|
|
ip_address=self._get_client_ip(),
|
|
user_agent=request.META.get('HTTP_USER_AGENT', '')[:255],
|
|
)
|
|
except Exception as e:
|
|
print(f"Failed to create audit log: {e}")
|
|
|
|
return response
|
|
|
|
|
|
class HTMXResponseMixin:
|
|
"""
|
|
Mixin to handle HTMX requests with partial template rendering.
|
|
|
|
Usage:
|
|
class MyView(HTMXResponseMixin, ListView):
|
|
template_name = 'myapp/list.html'
|
|
htmx_template_name = 'myapp/partials/list_partial.html'
|
|
"""
|
|
|
|
htmx_template_name = None # Override in subclass
|
|
|
|
def render_to_response(self, context, **response_kwargs):
|
|
"""Render partial template for HTMX requests."""
|
|
# Check if this is an HTMX request
|
|
if self.request.headers.get('HX-Request'):
|
|
# Use partial template if specified
|
|
if self.htmx_template_name:
|
|
return render(
|
|
self.request,
|
|
self.htmx_template_name,
|
|
context,
|
|
**response_kwargs
|
|
)
|
|
|
|
# Default rendering for non-HTMX requests
|
|
return super().render_to_response(context, **response_kwargs)
|
|
|
|
|
|
class ObjectOwnershipMixin:
|
|
"""
|
|
Mixin to check if user has permission to access/modify an object.
|
|
|
|
Checks:
|
|
1. Object belongs to user's tenant
|
|
2. User has appropriate role
|
|
3. Optional: User is the creator/owner
|
|
|
|
Usage:
|
|
class MyUpdateView(ObjectOwnershipMixin, UpdateView):
|
|
check_ownership = True # Require user to be owner
|
|
"""
|
|
|
|
check_ownership = False # Set to True to require ownership
|
|
ownership_field = 'created_by' # Field to check for ownership
|
|
|
|
def get_object(self, queryset=None):
|
|
"""Get object and verify permissions."""
|
|
obj = super().get_object(queryset)
|
|
|
|
# Check tenant
|
|
if hasattr(obj, 'tenant') and obj.tenant != self.request.user.tenant:
|
|
raise PermissionDenied("You don't have permission to access this object.")
|
|
|
|
# Check ownership if required
|
|
if self.check_ownership:
|
|
if hasattr(obj, self.ownership_field):
|
|
owner = getattr(obj, self.ownership_field)
|
|
if owner != self.request.user:
|
|
raise PermissionDenied("You don't have permission to modify this object.")
|
|
|
|
return obj
|
|
|
|
|
|
class SuccessMessageMixin:
|
|
"""
|
|
Mixin to add success messages after form submission.
|
|
|
|
Usage:
|
|
class MyCreateView(SuccessMessageMixin, CreateView):
|
|
success_message = "Object created successfully!"
|
|
"""
|
|
|
|
success_message = ""
|
|
|
|
def form_valid(self, form):
|
|
"""Add success message."""
|
|
response = super().form_valid(form)
|
|
|
|
if self.success_message:
|
|
from django.contrib import messages
|
|
messages.success(self.request, self.success_message)
|
|
|
|
return response
|
|
|
|
|
|
class PaginationMixin:
|
|
"""
|
|
Mixin to add consistent pagination across list views.
|
|
|
|
Usage:
|
|
class MyListView(PaginationMixin, ListView):
|
|
paginate_by = 25
|
|
"""
|
|
|
|
paginate_by = 25 # Default pagination
|
|
|
|
def get_context_data(self, **kwargs):
|
|
"""Add pagination context."""
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
# Add pagination info
|
|
if 'page_obj' in context:
|
|
page_obj = context['page_obj']
|
|
context['pagination_info'] = {
|
|
'current_page': page_obj.number,
|
|
'total_pages': page_obj.paginator.num_pages,
|
|
'total_items': page_obj.paginator.count,
|
|
'has_previous': page_obj.has_previous(),
|
|
'has_next': page_obj.has_next(),
|
|
'start_index': page_obj.start_index(),
|
|
'end_index': page_obj.end_index(),
|
|
}
|
|
|
|
return context
|
|
|
|
|
|
class ConsentRequiredMixin:
|
|
"""
|
|
Mixin to enforce consent verification before creating clinical documentation.
|
|
|
|
This mixin checks if a patient has signed the required consent forms before
|
|
allowing clinical documentation to be created. It should be used on all
|
|
clinical CreateView classes (consultations, sessions, assessments, etc.).
|
|
|
|
Attributes:
|
|
consent_service_type (str): Required. The service type to check consent for
|
|
(e.g., 'ABA', 'MEDICAL', 'OT', 'SLP').
|
|
consent_redirect_url (str): Optional. URL to redirect to if consent missing.
|
|
Defaults to consent creation page.
|
|
consent_error_message (str): Optional. Custom error message to display.
|
|
consent_skip_check (bool): Optional. Set to True to skip consent check
|
|
(useful for testing or special cases).
|
|
|
|
Methods:
|
|
get_patient(): Must be implemented by subclass to return the Patient instance.
|
|
|
|
Usage:
|
|
class ABAConsultCreateView(ConsentRequiredMixin, CreateView):
|
|
consent_service_type = 'ABA'
|
|
|
|
def get_patient(self):
|
|
patient_id = self.request.GET.get('patient')
|
|
return Patient.objects.get(pk=patient_id, tenant=self.request.user.tenant)
|
|
|
|
Example with custom error message:
|
|
class MedicalConsultCreateView(ConsentRequiredMixin, CreateView):
|
|
consent_service_type = 'MEDICAL'
|
|
consent_error_message = "Patient must sign medical consent before consultation."
|
|
|
|
def get_patient(self):
|
|
return self.get_object().patient
|
|
"""
|
|
|
|
consent_service_type = None
|
|
consent_redirect_url = None
|
|
consent_error_message = None
|
|
consent_skip_check = False
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
"""Check consent before allowing access to the view."""
|
|
|
|
# Skip check if explicitly disabled
|
|
if self.consent_skip_check:
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
# Validate configuration
|
|
if not self.consent_service_type:
|
|
raise ImproperlyConfigured(
|
|
f"{self.__class__.__name__} must define consent_service_type attribute. "
|
|
f"Set it to the service type (e.g., 'ABA', 'MEDICAL', 'OT', 'SLP')."
|
|
)
|
|
|
|
# Get patient
|
|
try:
|
|
patient = self.get_patient()
|
|
except Exception as e:
|
|
messages.error(
|
|
request,
|
|
f"Error retrieving patient information: {str(e)}. "
|
|
f"Please ensure patient is specified correctly."
|
|
)
|
|
return redirect('core:patient_list')
|
|
|
|
# Only check consent if patient is available
|
|
if patient:
|
|
# Import here to avoid circular imports
|
|
from core.services import ConsentService
|
|
|
|
# Verify consent
|
|
has_consent, consent_message = ConsentService.verify_consent_for_service(
|
|
patient,
|
|
self.consent_service_type
|
|
)
|
|
|
|
if not has_consent:
|
|
# Get missing consents for detailed feedback
|
|
missing_consents = ConsentService.get_missing_consents(
|
|
patient,
|
|
self.consent_service_type
|
|
)
|
|
|
|
# Build error message
|
|
if self.consent_error_message:
|
|
error_msg = self.consent_error_message
|
|
else:
|
|
error_msg = (
|
|
f"Cannot create {self.consent_service_type} documentation: {consent_message}. "
|
|
)
|
|
if missing_consents:
|
|
error_msg += f"Missing consent types: {', '.join(missing_consents)}."
|
|
|
|
messages.error(request, error_msg)
|
|
|
|
# Determine redirect URL
|
|
if self.consent_redirect_url:
|
|
redirect_url = self.consent_redirect_url
|
|
else:
|
|
# Default: redirect to patient detail page with consent tab
|
|
redirect_url = reverse('core:patient_detail', kwargs={'pk': patient.pk})
|
|
redirect_url += '?tab=consents&missing=' + ','.join(missing_consents)
|
|
|
|
return redirect(redirect_url)
|
|
|
|
# All checks passed, proceed with normal dispatch
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
def get_patient(self):
|
|
"""
|
|
Get the patient for consent verification.
|
|
|
|
This method must be implemented by the subclass to return the Patient
|
|
instance that needs consent verification.
|
|
|
|
Common implementations:
|
|
- Get from URL parameter: Patient.objects.get(pk=self.request.GET.get('patient'))
|
|
- Get from form: self.get_form().instance.patient
|
|
- Get from appointment: Appointment.objects.get(pk=...).patient
|
|
|
|
Returns:
|
|
Patient: The patient instance to check consent for
|
|
|
|
Raises:
|
|
NotImplementedError: If not implemented by subclass
|
|
"""
|
|
raise NotImplementedError(
|
|
f"{self.__class__.__name__} must implement get_patient() method. "
|
|
f"This method should return the Patient instance for consent verification."
|
|
)
|