Partner System Integration

   integration:
     name: "partner-referral-sync"
     partner: "Example Partner Organisation"
     version: "1.0"
     created: "2024-11-15"

   authentication:
     type: "oauth2_client_credentials"
     token_endpoint: "https://auth.partner.example.org/oauth/token"
     scopes:
       - "read:beneficiaries"
       - "write:referrals"
     token_expiry_seconds: 3600

   endpoints:
     submit_referral:
       method: "POST"
       url: "https://api.partner.example.org/v2/referrals"
       content_type: "application/json"
       rate_limit: "100/minute"

     get_referral_status:
       method: "GET"
       url: "https://api.partner.example.org/v2/referrals/{referral_id}"
       rate_limit: "300/minute"

   data_mapping:
     outbound:
       - source: "case.client_name"
         target: "beneficiary.full_name"
       - source: "case.date_of_birth"
         target: "beneficiary.dob"
         format: "YYYY-MM-DD"

   error_handling:
     retry_codes: [429, 500, 502, 503, 504]
     max_retries: 3
     backoff_seconds: [1, 5, 30]

   import hvac
   import requests
   from datetime import datetime, timedelta

   class PartnerIntegrationClient:
       def __init__(self, vault_path="secret/integrations/partner-example"):
           self.vault = hvac.Client()
           self.credentials = self.vault.secrets.kv.read_secret_version(
               path=vault_path
           )['data']['data']
           self.token = None
           self.token_expiry = None

       def get_access_token(self):
           if self.token and datetime.now() < self.token_expiry:
               return self.token

           response = requests.post(
               self.credentials['token_endpoint'],
               data={
                   'grant_type': 'client_credentials',
                   'client_id': self.credentials['client_id'],
                   'client_secret': self.credentials['client_secret'],
                   'scope': 'read:beneficiaries write:referrals'
               }
           )
           response.raise_for_status()
           token_data = response.json()

           self.token = token_data['access_token']
           # Refresh 5 minutes before expiry
           self.token_expiry = datetime.now() + timedelta(
               seconds=token_data['expires_in'] - 300
           )
           return self.token

   from datetime import datetime

   class ReferralTransformer:
       """Transform internal case data to partner referral format."""

       FIELD_MAPPING = {
           'case.client_name': 'beneficiary.full_name',
           'case.date_of_birth': 'beneficiary.dob',
           'case.gender': 'beneficiary.gender',
           'case.contact_phone': 'beneficiary.phone',
           'case.needs_assessment': 'referral.needs',
           'case.urgency': 'referral.priority',
       }

       GENDER_MAPPING = {
           'male': 'M',
           'female': 'F',
           'other': 'O',
           'prefer_not_to_say': 'U',
       }

       PRIORITY_MAPPING = {
           'critical': 1,
           'high': 2,
           'medium': 3,
           'low': 4,
       }

       def transform(self, case_data: dict) -> dict:
           """Transform case data to partner referral format."""
           return {
               'beneficiary': {
                   'full_name': case_data['client_name'],
                   'dob': self._format_date(case_data['date_of_birth']),
                   'gender': self.GENDER_MAPPING.get(
                       case_data.get('gender', '').lower(), 'U'
                   ),
                   'phone': self._format_phone(case_data.get('contact_phone')),
               },
               'referral': {
                   'source_system': 'your-org-case-management',
                   'source_id': case_data['case_id'],
                   'needs': case_data.get('needs_assessment', []),
                   'priority': self.PRIORITY_MAPPING.get(
                       case_data.get('urgency', 'medium'), 3
                   ),
                   'submitted_at': datetime.utcnow().isoformat() + 'Z',
               }
           }

       def _format_date(self, date_value) -> str:
           if isinstance(date_value, datetime):
               return date_value.strftime('%Y-%m-%d')
           return str(date_value)[:10]

       def _format_phone(self, phone: str) -> str:
           if not phone:
               return None
           # Remove non-digits, ensure international format
           digits = ''.join(c for c in phone if c.isdigit())
           if len(digits) == 10:
               return f'+1{digits}'  # Adjust country code as needed
           return f'+{digits}'

   import time
   import logging
   from requests.exceptions import RequestException

   logger = logging.getLogger(__name__)

   class PartnerAPIClient(PartnerIntegrationClient):
       RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
       MAX_RETRIES = 3
       BACKOFF_SECONDS = [1, 5, 30]

       def submit_referral(self, referral_data: dict) -> dict:
           """Submit referral with retry logic."""
           url = "https://api.partner.example.org/v2/referrals"

           for attempt in range(self.MAX_RETRIES + 1):
               try:
                   response = requests.post(
                       url,
                       json=referral_data,
                       headers={
                           'Authorization': f'Bearer {self.get_access_token()}',
                           'Content-Type': 'application/json',
                           'X-Request-ID': self._generate_request_id(),
                       },
                       timeout=30
                   )

                   if response.status_code == 201:
                       logger.info(f"Referral submitted: {response.json()['id']}")
                       return response.json()

                   if response.status_code in self.RETRY_STATUS_CODES:
                       if attempt < self.MAX_RETRIES:
                           wait_time = self.BACKOFF_SECONDS[attempt]
                           logger.warning(
                               f"Retryable error {response.status_code}, "
                               f"waiting {wait_time}s (attempt {attempt + 1})"
                           )
                           time.sleep(wait_time)
                           continue

                   # Non-retryable error
                   response.raise_for_status()

               except RequestException as e:
                   if attempt < self.MAX_RETRIES:
                       wait_time = self.BACKOFF_SECONDS[attempt]
                       logger.warning(f"Request failed: {e}, retrying in {wait_time}s")
                       time.sleep(wait_time)
                   else:
                       logger.error(f"Request failed after {self.MAX_RETRIES} retries")
                       raise

           raise Exception("Max retries exceeded")

   import hmac
   import hashlib
   from flask import Flask, request, jsonify

   app = Flask(__name__)
   WEBHOOK_SECRET = get_secret('partner-webhook-secret')

   @app.route('/webhooks/partner/referral-status', methods=['POST'])
   def handle_referral_status():
       # Validate webhook signature
       signature = request.headers.get('X-Webhook-Signature')
       if not validate_signature(request.data, signature):
           logger.warning("Invalid webhook signature")
           return jsonify({'error': 'Invalid signature'}), 401

       payload = request.json

       # Process status update
       referral_id = payload['referral_id']
       new_status = payload['status']

       # Update local record
       update_referral_status(
           partner_referral_id=referral_id,
           status=new_status,
           updated_at=payload['updated_at']
       )

       logger.info(f"Referral {referral_id} status updated to {new_status}")
       return jsonify({'received': True}), 200

   def validate_signature(payload: bytes, signature: str) -> bool:
       expected = hmac.new(
           WEBHOOK_SECRET.encode(),
           payload,
           hashlib.sha256
       ).hexdigest()
       return hmac.compare_digest(f'sha256={expected}', signature)

   import csv
   import io
   import paramiko
   from datetime import datetime

   class PartnerFileTransfer:
       def __init__(self, vault_path="secret/integrations/partner-sftp"):
           creds = get_vault_secret(vault_path)
           self.hostname = creds['hostname']
           self.username = creds['username']
           self.private_key = paramiko.Ed25519Key.from_private_key(
               io.StringIO(creds['private_key'])
           )
           self.remote_path = creds['remote_path']

       def generate_beneficiary_extract(self) -> str:
           """Generate CSV extract of beneficiaries for partner."""
           beneficiaries = get_beneficiaries_for_partner(
               partner_id='partner-example',
               since_last_sync=True
           )

           output = io.StringIO()
           writer = csv.DictWriter(output, fieldnames=[
               'source_id', 'full_name', 'dob', 'gender',
               'registration_date', 'location', 'status'
           ])
           writer.writeheader()

           for b in beneficiaries:
               writer.writerow({
                   'source_id': b['id'],
                   'full_name': b['name'],
                   'dob': b['date_of_birth'].strftime('%Y-%m-%d'),
                   'gender': b['gender'][0].upper() if b['gender'] else 'U',
                   'registration_date': b['registered_at'].strftime('%Y-%m-%d'),
                   'location': b['location_code'],
                   'status': b['status'],
               })

           return output.getvalue()

       def upload_file(self, content: str, filename: str):
           """Upload file to partner SFTP server."""
           transport = paramiko.Transport((self.hostname, 22))
           transport.connect(username=self.username, pkey=self.private_key)
           sftp = paramiko.SFTPClient.from_transport(transport)

           try:
               remote_file = f"{self.remote_path}/{filename}"
               with sftp.file(remote_file, 'w') as f:
                   f.write(content)
               logger.info(f"Uploaded {filename} to {self.hostname}")
           finally:
               sftp.close()
               transport.close()

       def run_daily_sync(self):
           """Execute daily file sync to partner."""
           timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
           filename = f"beneficiaries_{timestamp}.csv"

           content = self.generate_beneficiary_extract()
           record_count = content.count('\n') - 1  # Subtract header

           self.upload_file(content, filename)

           # Log sync for audit
           log_sync_event(
               partner='partner-example',
               filename=filename,
               records=record_count,
               direction='outbound'
           )

   class PartnerFileReceiver(PartnerFileTransfer):
       def __init__(self):
           super().__init__()
           self.incoming_path = "/incoming/from-partner"
           self.processed_path = "/processed/from-partner"

       def check_for_files(self) -> list:
           """Check partner SFTP for new files."""
           transport = paramiko.Transport((self.hostname, 22))
           transport.connect(username=self.username, pkey=self.private_key)
           sftp = paramiko.SFTPClient.from_transport(transport)

           try:
               files = sftp.listdir(self.incoming_path)
               return [f for f in files if f.endswith('.csv')]
           finally:
               sftp.close()
               transport.close()

       def download_and_process(self, filename: str):
           """Download file from partner and process contents."""
           transport = paramiko.Transport((self.hostname, 22))
           transport.connect(username=self.username, pkey=self.private_key)
           sftp = paramiko.SFTPClient.from_transport(transport)

           try:
               remote_path = f"{self.incoming_path}/{filename}"
               with sftp.file(remote_path, 'r') as f:
                   content = f.read().decode('utf-8')

               # Process the file
               records_processed = self.process_partner_file(content, filename)

               # Move to processed folder
               processed_path = f"{self.processed_path}/{filename}"
               sftp.rename(remote_path, processed_path)

               log_sync_event(
                   partner='partner-example',
                   filename=filename,
                   records=records_processed,
                   direction='inbound'
               )
           finally:
               sftp.close()
               transport.close()

   federation:
     partner: "Example Partner Organisation"
     type: "saml"
     direction: "your-idp-to-partner-sp"
     established: "2024-11-15"

     technical:
       your_entity_id: "https://idp.yourorg.example.org/realms/your-realm"
       partner_entity_id: "https://app.partner.example.org"
       partner_acs: "https://app.partner.example.org/saml/acs"
       signing_cert_expiry: "2026-11-15"

     access_control:
       group: "partner-example-users"
       approver: "IT Manager"
       review_frequency: "quarterly"

     attributes_shared:
       - email
       - display_name
       - department

     contacts:
       partner_technical: "techsupport@partner.example.org"
       your_technical: "it@yourorg.example.org"

groups:
- name: partner-integration
  rules:
  - alert: PartnerIntegrationAuthFailure
    expr: |
      sum(rate(partner_api_auth_failures_total[5m])) > 0.1
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Partner API authentication failures detected"
      description: "{{ $value }} auth failures/sec to partner API"

  - alert: PartnerIntegrationHighErrorRate
    expr: |
      sum(rate(partner_api_requests_total{status=~"5.."}[10m]))
      / sum(rate(partner_api_requests_total[10m])) > 0.05
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "Partner API error rate above 5%"

  - alert: PartnerFileSyncMissed
    expr: |
      time() - partner_file_sync_last_success_timestamp > 90000
    labels:
      severity: critical
    annotations:
      summary: "Partner file sync has not completed in 25 hours"

  - alert: PartnerWebhookBacklog
    expr: |
      partner_webhook_queue_depth > 100
    for: 15m
    labels:
      severity: warning
    annotations:
      summary: "Partner webhook processing backlog"