523 lines
17 KiB
Python
523 lines
17 KiB
Python
"""
|
|
Integration app forms for CRUD operations.
|
|
"""
|
|
|
|
from django import forms
|
|
from django.core.exceptions import ValidationError
|
|
from django.contrib.auth import get_user_model
|
|
import json
|
|
import requests
|
|
|
|
from .models import (
|
|
ExternalSystem, IntegrationEndpoint, DataMapping, WebhookEndpoint
|
|
)
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class ExternalSystemForm(forms.ModelForm):
|
|
"""
|
|
Form for creating and updating external systems.
|
|
"""
|
|
|
|
class Meta:
|
|
model = ExternalSystem
|
|
fields = [
|
|
'name', 'system_type', 'description', 'base_url',
|
|
'authentication_type', 'authentication_config', 'timeout_seconds',
|
|
'retry_attempts', 'is_active'
|
|
]
|
|
widgets = {
|
|
'name': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Enter system name'
|
|
}),
|
|
'system_type': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'description': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 3,
|
|
'placeholder': 'Enter system description'
|
|
}),
|
|
'base_url': forms.URLInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'https://api.example.com'
|
|
}),
|
|
'authentication_type': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'authentication_config': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 5,
|
|
'placeholder': '{"api_key": "your_api_key", "username": "user", "password": "pass"}'
|
|
}),
|
|
'timeout_seconds': forms.NumberInput(attrs={
|
|
'class': 'form-control',
|
|
'min': 1,
|
|
'max': 300,
|
|
'value': 30
|
|
}),
|
|
'retry_attempts': forms.NumberInput(attrs={
|
|
'class': 'form-control',
|
|
'min': 0,
|
|
'max': 10,
|
|
'value': 3
|
|
}),
|
|
'is_active': forms.CheckboxInput(attrs={
|
|
'class': 'form-check-input'
|
|
}),
|
|
}
|
|
|
|
def clean_authentication_config(self):
|
|
"""
|
|
Validate authentication configuration JSON.
|
|
"""
|
|
auth_config = self.cleaned_data.get('authentication_config')
|
|
if auth_config:
|
|
try:
|
|
json.loads(auth_config)
|
|
except json.JSONDecodeError:
|
|
raise ValidationError('Authentication configuration must be valid JSON.')
|
|
return auth_config
|
|
|
|
def clean_base_url(self):
|
|
"""
|
|
Validate base URL format.
|
|
"""
|
|
base_url = self.cleaned_data.get('base_url')
|
|
if base_url and not base_url.startswith(('http://', 'https://')):
|
|
raise ValidationError('Base URL must start with http:// or https://')
|
|
return base_url
|
|
|
|
|
|
class IntegrationEndpointForm(forms.ModelForm):
|
|
"""
|
|
Form for creating and updating integration endpoints.
|
|
"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.user = kwargs.pop('user', None)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
if self.user:
|
|
self.fields['external_system'].queryset = ExternalSystem.objects.filter(
|
|
tenant=self.user.tenant,
|
|
is_active=True
|
|
).order_by('name')
|
|
|
|
class Meta:
|
|
model = IntegrationEndpoint
|
|
fields = [
|
|
'external_system', 'name', 'description', 'path',
|
|
'method', 'headers', 'request_mapping', 'response_mapping',
|
|
'endpoint_type', 'direction', 'is_active'
|
|
]
|
|
widgets = {
|
|
'external_system': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'name': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Enter endpoint name'
|
|
}),
|
|
'description': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 3,
|
|
'placeholder': 'Enter endpoint description'
|
|
}),
|
|
'path': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': '/api/v1/patients'
|
|
}),
|
|
'method': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'headers': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 4,
|
|
'placeholder': '{"Content-Type": "application/json", "Authorization": "Bearer {token}"}'
|
|
}),
|
|
'request_mapping': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 6,
|
|
'placeholder': '{"patient_id": "{patient_id}", "data": "{data}"}'
|
|
}),
|
|
'response_mapping': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 6,
|
|
'placeholder': '{"patient_id": "$.data.id", "status": "$.status"}'
|
|
}),
|
|
'endpoint_type': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'direction': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'is_active': forms.CheckboxInput(attrs={
|
|
'class': 'form-check-input'
|
|
}),
|
|
}
|
|
|
|
def clean_headers(self):
|
|
"""
|
|
Validate headers JSON.
|
|
"""
|
|
headers = self.cleaned_data.get('headers')
|
|
if headers:
|
|
try:
|
|
json.loads(headers)
|
|
except json.JSONDecodeError:
|
|
raise ValidationError('Headers must be valid JSON.')
|
|
return headers
|
|
|
|
def clean_request_template(self):
|
|
"""
|
|
Validate request template JSON.
|
|
"""
|
|
request_template = self.cleaned_data.get('request_template')
|
|
if request_template:
|
|
try:
|
|
json.loads(request_template)
|
|
except json.JSONDecodeError:
|
|
raise ValidationError('Request template must be valid JSON.')
|
|
return request_template
|
|
|
|
def clean_response_mapping(self):
|
|
"""
|
|
Validate response mapping JSON.
|
|
"""
|
|
response_mapping = self.cleaned_data.get('response_mapping')
|
|
if response_mapping:
|
|
try:
|
|
json.loads(response_mapping)
|
|
except json.JSONDecodeError:
|
|
raise ValidationError('Response mapping must be valid JSON.')
|
|
return response_mapping
|
|
|
|
|
|
class DataMappingForm(forms.ModelForm):
|
|
"""
|
|
Form for creating and updating data mappings.
|
|
"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.user = kwargs.pop('user', None)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
if self.user:
|
|
self.fields['endpoint'].queryset = IntegrationEndpoint.objects.filter(
|
|
external_system__tenant=self.user.tenant,
|
|
is_active=True
|
|
).order_by('name')
|
|
|
|
class Meta:
|
|
model = DataMapping
|
|
fields = [
|
|
'endpoint', 'name', 'source_field', 'target_field',
|
|
'mapping_type', 'transformation_type', 'transformation_config',
|
|
'is_active'
|
|
]
|
|
widgets = {
|
|
'endpoint': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'name': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Enter mapping name'
|
|
}),
|
|
'source_field': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'source.field.path'
|
|
}),
|
|
'target_field': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'target.field.path'
|
|
}),
|
|
'mapping_type': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'transformation_type': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'transformation_config': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 4,
|
|
'placeholder': 'Enter transformation configuration (JSON format)'
|
|
}),
|
|
'is_active': forms.CheckboxInput(attrs={
|
|
'class': 'form-check-input'
|
|
}),
|
|
}
|
|
|
|
def clean(self):
|
|
"""
|
|
Validate field mapping consistency.
|
|
"""
|
|
cleaned_data = super().clean()
|
|
source_field = cleaned_data.get('source_field')
|
|
target_field = cleaned_data.get('target_field')
|
|
direction = cleaned_data.get('direction')
|
|
|
|
if direction == 'INBOUND' and not source_field:
|
|
raise ValidationError('Source field is required for inbound mappings.')
|
|
|
|
if direction == 'OUTBOUND' and not target_field:
|
|
raise ValidationError('Target field is required for outbound mappings.')
|
|
|
|
return cleaned_data
|
|
|
|
|
|
class WebhookEndpointForm(forms.ModelForm):
|
|
"""
|
|
Form for creating and updating webhook endpoints.
|
|
"""
|
|
|
|
class Meta:
|
|
model = WebhookEndpoint
|
|
fields = [
|
|
'name', 'description', 'url_path', 'allowed_methods',
|
|
'authentication_type', 'authentication_config', 'processing_config',
|
|
'rate_limit_per_minute', 'is_active'
|
|
]
|
|
widgets = {
|
|
'name': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Enter webhook endpoint name'
|
|
}),
|
|
'description': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 3,
|
|
'placeholder': 'Enter webhook description'
|
|
}),
|
|
'url_path': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': '/webhook/endpoint/path'
|
|
}),
|
|
'allowed_methods': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 2,
|
|
'placeholder': '["POST", "PUT"]'
|
|
}),
|
|
'authentication_type': forms.Select(attrs={
|
|
'class': 'form-control'
|
|
}),
|
|
'authentication_config': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 4,
|
|
'placeholder': '{"api_key": "your_key", "secret": "your_secret"}'
|
|
}),
|
|
'processing_config': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 6,
|
|
'placeholder': '{"event": "{event_type}", "data": "{data}", "timestamp": "{timestamp}"}'
|
|
}),
|
|
'rate_limit_per_minute': forms.NumberInput(attrs={
|
|
'class': 'form-control',
|
|
'min': 1,
|
|
'max': 1000
|
|
}),
|
|
'is_active': forms.CheckboxInput(attrs={
|
|
'class': 'form-check-input'
|
|
}),
|
|
}
|
|
|
|
def clean_headers(self):
|
|
"""
|
|
Validate headers JSON.
|
|
"""
|
|
headers = self.cleaned_data.get('headers')
|
|
if headers:
|
|
try:
|
|
json.loads(headers)
|
|
except json.JSONDecodeError:
|
|
raise ValidationError('Headers must be valid JSON.')
|
|
return headers
|
|
|
|
def clean_payload_template(self):
|
|
"""
|
|
Validate payload template JSON.
|
|
"""
|
|
payload_template = self.cleaned_data.get('payload_template')
|
|
if payload_template:
|
|
try:
|
|
json.loads(payload_template)
|
|
except json.JSONDecodeError:
|
|
raise ValidationError('Payload template must be valid JSON.')
|
|
return payload_template
|
|
|
|
def clean_webhook_url(self):
|
|
"""
|
|
Validate webhook URL format.
|
|
"""
|
|
webhook_url = self.cleaned_data.get('webhook_url')
|
|
if webhook_url and not webhook_url.startswith(('http://', 'https://')):
|
|
raise ValidationError('Webhook URL must start with http:// or https://')
|
|
return webhook_url
|
|
|
|
def clean_allowed_ips(self):
|
|
"""
|
|
Validate IP addresses format.
|
|
"""
|
|
allowed_ips = self.cleaned_data.get('allowed_ips')
|
|
if allowed_ips:
|
|
import ipaddress
|
|
ip_list = [ip.strip() for ip in allowed_ips.split('\n') if ip.strip()]
|
|
for ip in ip_list:
|
|
try:
|
|
ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
try:
|
|
ipaddress.ip_network(ip, strict=False)
|
|
except ValueError:
|
|
raise ValidationError(f'Invalid IP address or network: {ip}')
|
|
return allowed_ips
|
|
|
|
|
|
# ============================================================================
|
|
# BULK OPERATION FORMS
|
|
# ============================================================================
|
|
|
|
class BulkEndpointExecutionForm(forms.Form):
|
|
"""
|
|
Form for bulk endpoint execution.
|
|
"""
|
|
endpoint_ids = forms.ModelMultipleChoiceField(
|
|
queryset=IntegrationEndpoint.objects.none(),
|
|
widget=forms.CheckboxSelectMultiple(attrs={
|
|
'class': 'form-check-input'
|
|
}),
|
|
label='Select Endpoints to Execute'
|
|
)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
user = kwargs.pop('user', None)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
if user:
|
|
self.fields['endpoint_ids'].queryset = IntegrationEndpoint.objects.filter(
|
|
tenant=user.tenant,
|
|
is_active=True
|
|
).order_by('endpoint_name')
|
|
|
|
|
|
class SystemHealthCheckForm(forms.Form):
|
|
"""
|
|
Form for system health check operations.
|
|
"""
|
|
system_ids = forms.ModelMultipleChoiceField(
|
|
queryset=ExternalSystem.objects.none(),
|
|
widget=forms.CheckboxSelectMultiple(attrs={
|
|
'class': 'form-check-input'
|
|
}),
|
|
label='Select Systems to Check'
|
|
)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
user = kwargs.pop('user', None)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
if user:
|
|
self.fields['system_ids'].queryset = ExternalSystem.objects.filter(
|
|
tenant=user.tenant,
|
|
is_active=True
|
|
).order_by('system_name')
|
|
|
|
|
|
# ============================================================================
|
|
# SEARCH AND FILTER FORMS
|
|
# ============================================================================
|
|
|
|
class IntegrationSearchForm(forms.Form):
|
|
"""
|
|
Form for searching across integration components.
|
|
"""
|
|
search = forms.CharField(
|
|
max_length=255,
|
|
required=False,
|
|
widget=forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Search systems, endpoints, mappings...'
|
|
})
|
|
)
|
|
|
|
component_type = forms.ChoiceField(
|
|
choices=[
|
|
('', 'All Components'),
|
|
('system', 'External Systems'),
|
|
('endpoint', 'Endpoints'),
|
|
('mapping', 'Data Mappings'),
|
|
('webhook', 'Webhooks'),
|
|
],
|
|
required=False,
|
|
widget=forms.Select(attrs={
|
|
'class': 'form-control'
|
|
})
|
|
)
|
|
|
|
status = forms.ChoiceField(
|
|
choices=[
|
|
('', 'All Statuses'),
|
|
('active', 'Active'),
|
|
('inactive', 'Inactive'),
|
|
],
|
|
required=False,
|
|
widget=forms.Select(attrs={
|
|
'class': 'form-control'
|
|
})
|
|
)
|
|
|
|
|
|
class ExecutionFilterForm(forms.Form):
|
|
"""
|
|
Form for filtering integration executions.
|
|
"""
|
|
start_date = forms.DateField(
|
|
required=False,
|
|
widget=forms.DateInput(attrs={
|
|
'class': 'form-control',
|
|
'type': 'date'
|
|
})
|
|
)
|
|
|
|
end_date = forms.DateField(
|
|
required=False,
|
|
widget=forms.DateInput(attrs={
|
|
'class': 'form-control',
|
|
'type': 'date'
|
|
})
|
|
)
|
|
|
|
status = forms.ChoiceField(
|
|
choices=[
|
|
('', 'All Statuses'),
|
|
('SUCCESS', 'Success'),
|
|
('FAILED', 'Failed'),
|
|
('RUNNING', 'Running'),
|
|
('CANCELLED', 'Cancelled'),
|
|
],
|
|
required=False,
|
|
widget=forms.Select(attrs={
|
|
'class': 'form-control'
|
|
})
|
|
)
|
|
|
|
endpoint = forms.ModelChoiceField(
|
|
queryset=IntegrationEndpoint.objects.none(),
|
|
required=False,
|
|
empty_label='All Endpoints',
|
|
widget=forms.Select(attrs={
|
|
'class': 'form-control'
|
|
})
|
|
)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
user = kwargs.pop('user', None)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
if user:
|
|
self.fields['endpoint'].queryset = IntegrationEndpoint.objects.filter(
|
|
tenant=user.tenant
|
|
).order_by('endpoint_name')
|
|
|