Skip to main content

API Integration Patterns

This page provides code examples and patterns for integrating the Triodian Metamodel with OpenMetadata. Examples are provided in Python and can be adapted for other languages.

Authentication Setup​

OpenMetadata Authentication​

import requests
from typing import Dict, Any

class OpenMetadataClient:
    """Client for OpenMetadata API interactions."""
    
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip('/')
        self.headers = {
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        }
    
    def get(self, endpoint: str) -> Dict[str, Any]:
        response = requests.get(
            f'{self.base_url}/api/v1/{endpoint}',
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()
    
    def post(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
        response = requests.post(
            f'{self.base_url}/api/v1/{endpoint}',
            headers=self.headers,
            json=data
        )
        response.raise_for_status()
        return response.json()
    
    def patch(self, endpoint: str, data: list) -> Dict[str, Any]:
        response = requests.patch(
            f'{self.base_url}/api/v1/{endpoint}',
            headers=self.headers,
            json=data
        )
        response.raise_for_status()
        return response.json()

# Initialize client
om_client = OpenMetadataClient(
    base_url='https://your-openmetadata-instance.com',
    api_key='your-api-key'
)

Pattern 1: Bulk Import Business Context​

Import Capabilities as Custom Properties​

import json
from typing import List, Dict

def load_triodian_records(filepath: str) -> List[Dict]:
    """Load a JSON array of Triodian records (e.g. capability records)."""
    with open(filepath, 'r') as f:
        return json.load(f)

def create_capability_custom_property_type(client: OpenMetadataClient):
    """Create custom property type for business capability."""
    custom_property = {
        "name": "capabilityID",
        "description": "Triodian capabilityID of the capability this asset supports",
        "propertyType": {
            "id": "string-type-id",  # Get from OM types endpoint
            "type": "type"
        }
    }
    
    # Apply to Table entity type
    client.patch(
        'metadata/types/name/table',
        [
            {
                "op": "add",
                "path": "/customProperties/-",
                "value": custom_property
            }
        ]
    )

def apply_capability_to_tables(
    client: OpenMetadataClient,
    capability: Dict,
    table_fqns: List[str]
):
    """Apply capability context to OpenMetadata tables."""
    for table_fqn in table_fqns:
        client.patch(
            f'tables/name/{table_fqn}',
            [
                {
                    "op": "add",
                    "path": "/extension/capabilityID",
                    "value": capability['capabilityID']
                },
                {
                    "op": "add",
                    "path": "/extension/capabilityTitle",
                    "value": capability['title']
                },
                {
                    "op": "add",
                    "path": "/extension/capabilityOwner",
                    "value": capability['owner']  # required in capability.schema.json
                }
            ]
        )

# Usage
# Capability records do not list data assets, so keep a separate mapping
# of capabilityID -> table FQNs, e.g. {"CAP-CRM-001": ["warehouse.crm.customers"]}
capabilities = load_triodian_records('capabilities.json')
with open('capability-asset-mapping.json') as f:
    asset_mapping = json.load(f)

for cap in capabilities:
    tables = asset_mapping.get(cap['capabilityID'], [])
    if tables:
        apply_capability_to_tables(om_client, cap, tables)

Pattern 2: Sync Stakeholders to Owners​

Map Triodian Stakeholders to OpenMetadata Users​

from typing import Optional

def map_stakeholder_to_user(
    client: OpenMetadataClient,
    stakeholder: Dict,
    email_directory: Dict[str, str]
) -> Optional[str]:
    """Map Triodian stakeholder to OpenMetadata user.

    Stakeholder records carry no contact details, so e-mail addresses come
    from an identity directory keyed by stakeholderID.
    """
    
    # Try to find existing user by email
    email = email_directory.get(stakeholder['stakeholderID'])
    if email:
        try:
            user = client.get(f'users/name/{email.split("@")[0]}')
            return user['id']
        except requests.HTTPError:
            pass
    
    # Create user if not exists
    user_data = {
        "name": stakeholder['stakeholderID'].lower().replace('-', '_'),
        "displayName": stakeholder['title'],
        "email": email or f"{stakeholder['stakeholderID']}@placeholder.com",
        "description": stakeholder.get('description', ''),
        "isAdmin": False
    }
    
    # Add custom properties for business context
    user_data['extension'] = {
        'stakeholderID': stakeholder['stakeholderID'],
        'stakeholderRole': stakeholder.get('role', '')
    }
    
    result = client.post('users', user_data)
    return result['id']

def assign_table_owner(
    client: OpenMetadataClient,
    table_fqn: str,
    owner_id: str
):
    """Assign owner to OpenMetadata table."""
    client.patch(
        f'tables/name/{table_fqn}',
        [
            {
                "op": "add",
                "path": "/owner",
                "value": {
                    "id": owner_id,
                    "type": "user"
                }
            }
        ]
    )

# Usage
stakeholders = load_triodian_records('stakeholders.json')
with open('email-directory.json') as f:
    email_directory = json.load(f)          # stakeholderID -> e-mail
with open('stakeholder-asset-mapping.json') as f:
    owned_assets = json.load(f)             # stakeholderID -> table FQNs

for stakeholder in stakeholders:
    user_id = map_stakeholder_to_user(om_client, stakeholder, email_directory)
    
    # Assign ownership of the assets this stakeholder is accountable for
    for asset_fqn in owned_assets.get(stakeholder['stakeholderID'], []):
        assign_table_owner(om_client, asset_fqn, user_id)

Pattern 3: Create Glossary from Domains​

Export Domain Definitions to OpenMetadata Glossary​

def create_glossary(client: OpenMetadataClient, name: str, description: str) -> str:
    """Create a glossary for business architecture terms."""
    glossary_data = {
        "name": name,
        "displayName": name.replace('-', ' ').title(),
        "description": description,
        "reviewers": [],
        "owner": None
    }
    
    result = client.post('glossaries', glossary_data)
    return result['id']

def create_glossary_term(
    client: OpenMetadataClient,
    glossary_id: str,
    domain_entity: Dict,
    domain_name: str
) -> str:
    """Create glossary term from Triodian domain entity."""
    term_data = {
        "glossary": {
            "id": glossary_id,
            "type": "glossary"
        },
        "name": domain_entity['title'].lower().replace(' ', '-'),
        "displayName": domain_entity['title'],
        "description": domain_entity.get('description', ''),
        "tags": [
            {
                "tagFQN": f"TriodianDomain.{domain_name}",
                "source": "Classification"
            }
        ]
    }
    
    # Keep the Triodian identifier (capabilityID, valueStreamID, ...) for traceability
    id_field = next((k for k in domain_entity if k.endswith('ID')), None)
    if id_field:
        term_data['extension'] = {id_field: domain_entity[id_field]}
    
    result = client.post('glossaryTerms', term_data)
    return result['id']

def export_domain_to_glossary(
    client: OpenMetadataClient,
    domain_data: List[Dict],
    domain_name: str
):
    """Export entire Triodian domain to OpenMetadata glossary."""
    
    # Create glossary for domain
    glossary_id = create_glossary(
        client,
        f"triodian-{domain_name}",
        f"Business architecture terms from Triodian {domain_name} domain"
    )
    
    # Create terms for each record
    for entity in domain_data:
        create_glossary_term(client, glossary_id, entity, domain_name)

# Usage: one JSON array of records per domain, named by domain type
domains = ['capability', 'valueStream', 'stakeholder', 'information']
for domain in domains:
    domain_data = load_triodian_records(f'{domain}.json')
    export_domain_to_glossary(om_client, domain_data, domain)

Pattern 4: Policy Synchronization​

Translate Business Policies to Governance Rules​

def translate_policy_to_om_rule(policy: Dict, om_policy_type: str = 'Lifecycle') -> Dict:
    """Translate Triodian policy to OpenMetadata policy.

    Triodian policies have no policy type, so the caller chooses the
    OpenMetadata policy type (e.g. from the policy's scope).
    """
    
    om_policy = {
        "name": policy['policyID'].lower().replace('-', '_'),
        "displayName": policy.get('title', policy['policyID']),
        "description": policy.get('description', ''),
        "policyType": om_policy_type,
        "enabled": True,
        "rules": []
    }
    
    # Translate the requirements text to rules
    rule = translate_requirements_to_rule(policy)
    if rule:
        om_policy['rules'].append(rule)
    
    # Add business context as extension
    om_policy['extension'] = {
        'policyID': policy['policyID'],
        'policyOwner': policy.get('owner', 'Unknown'),
        'enforcedBy': policy.get('enforcedBy', ''),
        'reviewFrequency': policy.get('reviewFrequency', '')
    }
    
    return om_policy

def translate_requirements_to_rule(policy: Dict) -> Optional[Dict]:
    """Translate a policy's requirements (free text) to an OpenMetadata rule."""
    import re
    requirements = policy.get('requirements', '')
    
    # Example: retention requirement such as "Delete customer PII after 730 days"
    days_match = re.search(r'(\d+)\s*days?', requirements)
    if days_match and 'delete' in requirements.lower():
        days = int(days_match.group(1))
        return {
            "name": f"{policy['policyID']}-retention",
            "description": requirements,
            "effect": "deny",
            "operations": ["All"],
            "resources": ["table"],
            "condition": f"matchAnyTag('PII') && daysSinceCreated > {days}"
        }
    
    # Add more rule translations as needed
    return None

def sync_policies(client: OpenMetadataClient, policies: List[Dict]):
    """Sync Triodian policies to OpenMetadata."""
    for policy in policies:
        om_policy = translate_policy_to_om_rule(policy)
        
        try:
            # Try to update existing policy
            existing = client.get(f'policies/name/{om_policy["name"]}')
            client.patch(
                f'policies/{existing["id"]}',
                [{"op": "replace", "path": "/rules", "value": om_policy['rules']}]
            )
        except requests.HTTPError:
            # Create new policy
            client.post('policies', om_policy)

# Usage
policies = load_triodian_records('policies.json')
sync_policies(om_client, policies)

Pattern 5: Value Stream to Pipeline Tagging​

Tag Pipelines with Value Stream Context​

def tag_pipeline_with_value_stream(
    client: OpenMetadataClient,
    pipeline_fqn: str,
    value_stream: Dict,
    stage: Optional[Dict] = None
):
    """Add value stream (and optionally stage) context to an OpenMetadata pipeline."""
    
    extension_data = {
        'valueStreamID': value_stream['valueStreamID'],
        'valueStreamTitle': value_stream.get('title', ''),
        'valueStreamPurpose': value_stream.get('purpose', '')
    }
    
    if stage:
        # A Value Stream Stage record (valueStreamStage)
        extension_data['stageID'] = stage['stageID']
        extension_data['stageTitle'] = stage.get('stageTitle', '')
        if stage.get('enablingCapabilityID'):
            extension_data['enablingCapabilityID'] = stage['enablingCapabilityID']
    
    # Build patch operations
    operations = [
        {
            "op": "add",
            "path": f"/extension/{key}",
            "value": value
        }
        for key, value in extension_data.items()
    ]
    
    client.patch(f'pipelines/name/{pipeline_fqn}', operations)

def create_value_stream_tag(client: OpenMetadataClient, value_stream: Dict):
    """Create classification tag for value stream."""
    
    # Ensure classification exists
    try:
        client.get('classifications/name/ValueStream')
    except requests.HTTPError:
        client.post('classifications', {
            "name": "ValueStream",
            "description": "Business value stream classification"
        })
    
    # Create tag for specific value stream
    tag_data = {
        "classification": "ValueStream",
        "name": value_stream['valueStreamID'],
        "description": value_stream.get('title', value_stream['valueStreamID'])
    }
    
    try:
        client.post('tags', tag_data)
    except requests.HTTPError:
        pass  # Tag already exists

# Usage
value_streams = load_triodian_records('value-streams.json')
stages = {s['stageID']: s for s in load_triodian_records('value-stream-stages.json')}
# e.g. [{"pipelineFQN": "airflow.order_processing", "valueStreamID": "VS-ORDER-001", "stageID": "VSS-ORDER-01"}]
pipeline_mapping = load_triodian_records('pipeline-vs-mapping.json')

for vs in value_streams:
    create_value_stream_tag(om_client, vs)
    
    for mapping in pipeline_mapping:
        if mapping['valueStreamID'] == vs['valueStreamID']:
            tag_pipeline_with_value_stream(
                om_client,
                mapping['pipelineFQN'],
                vs,
                stages.get(mapping.get('stageID'))
            )

Pattern 6: Bi-Directional Sync Service​

Event-Driven Synchronization​

from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class TriodianOMSync:
    """Bi-directional sync service between Triodian and OpenMetadata."""
    
    def __init__(
        self,
        om_client: OpenMetadataClient,
        triodian_api_url: str,
        triodian_api_key: str,
        asset_mapping: Dict[str, List[str]]
    ):
        self.om_client = om_client
        # Your repository of Triodian records (any service that stores them)
        self.triodian_url = triodian_api_url
        self.triodian_headers = {'Authorization': f'Bearer {triodian_api_key}'}
        # capabilityID -> table FQNs
        self.asset_mapping = asset_mapping
    
    def sync_capability_to_om(self, capability: Dict):
        """Sync capability update to OpenMetadata."""
        logger.info(f"Syncing capability {capability['capabilityID']} to OM")
        
        # Update custom properties on related tables
        for asset in self.asset_mapping.get(capability['capabilityID'], []):
            try:
                self.om_client.patch(
                    f'tables/name/{asset}',
                    [
                        {
                            "op": "add",
                            "path": "/extension/capabilityID",
                            "value": capability['capabilityID']
                        },
                        {
                            "op": "add",
                            "path": "/extension/capabilityTitle",
                            "value": capability['title']
                        },
                        {
                            "op": "add",
                            "path": "/extension/lastSyncedAt",
                            "value": datetime.utcnow().isoformat()
                        }
                    ]
                )
                logger.info(f"Updated table {asset} with capability context")
            except Exception as e:
                logger.error(f"Failed to update table {asset}: {e}")
    
    def sync_ownership_to_triodian(self, table_fqn: str, owner_id: str):
        """Sync ownership change from OM to Triodian."""
        logger.info(f"Syncing ownership change for {table_fqn} to Triodian")
        
        # Get table details including the capability it supports
        table = self.om_client.get(f'tables/name/{table_fqn}')
        capability_id = table.get('extension', {}).get('capabilityID')
        
        if capability_id:
            # Update the capability's owner (a field defined in capability.schema.json)
            owner = self.om_client.get(f'users/{owner_id}')
            try:
                response = requests.patch(
                    f'{self.triodian_url}/capabilities/{capability_id}',
                    headers=self.triodian_headers,
                    json={'owner': owner.get('displayName', owner['name'])}
                )
                response.raise_for_status()
                logger.info(f"Updated Triodian capability {capability_id}")
            except Exception as e:
                logger.error(f"Failed to update Triodian: {e}")
    
    def handle_om_event(self, event: Dict):
        """Handle OpenMetadata webhook event."""
        event_type = event.get('eventType')
        entity = event.get('entity', {})
        
        if event_type == 'entityUpdated' and entity.get('entityType') == 'table':
            # Check if ownership changed
            change_description = event.get('changeDescription', {})
            for field in change_description.get('fieldsUpdated', []):
                if field.get('name') == 'owner':
                    new_owner = field.get('newValue', {}).get('id')
                    if new_owner:
                        self.sync_ownership_to_triodian(
                            entity['fullyQualifiedName'],
                            new_owner
                        )
    
    def run_full_sync(self):
        """Run full synchronization."""
        logger.info("Starting full sync")
        
        # Sync capabilities to OM
        capabilities = self._load_all_capabilities()
        for cap in capabilities:
            self.sync_capability_to_om(cap)
        
        logger.info("Full sync completed")
    
    def _load_all_capabilities(self) -> List[Dict]:
        """Load all capabilities from Triodian."""
        response = requests.get(
            f'{self.triodian_url}/capabilities',
            headers=self.triodian_headers
        )
        response.raise_for_status()
        return response.json()  # a JSON array of capability records

# Usage
sync_service = TriodianOMSync(
    om_client=om_client,
    triodian_api_url='https://your-architecture-repository.example.com/api',
    triodian_api_key='your-repository-key',
    asset_mapping=asset_mapping
)

# Run initial sync
sync_service.run_full_sync()

# Handle webhook events (in your webhook handler)
# sync_service.handle_om_event(webhook_payload)

Error Handling Best Practices​

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def safe_api_call(client: OpenMetadataClient, method: str, *args, **kwargs):
    """Execute API call with retry logic."""
    func = getattr(client, method)
    return func(*args, **kwargs)

def batch_update_with_error_handling(
    client: OpenMetadataClient,
    updates: List[Dict]
) -> Dict[str, List]:
    """Batch update with error tracking."""
    results = {'success': [], 'failed': []}
    
    for update in updates:
        try:
            safe_api_call(
                client,
                'patch',
                update['endpoint'],
                update['operations']
            )
            results['success'].append(update['endpoint'])
        except Exception as e:
            results['failed'].append({
                'endpoint': update['endpoint'],
                'error': str(e)
            })
            logger.error(f"Failed to update {update['endpoint']}: {e}")
    
    return results
Last updated on