Development and Custom Transforms

Error Handling

Modified on: Tue, 14 Jul, 2026 at 11:35 AM

Proper error handling ensures your transforms provide meaningful feedback to users when things go wrong. This guide covers the Maltego exception hierarchy and best practices.

Exception Types

MaltegoException

The base exception for user-visible errors. Use when you want users to see a specific error message:

from maltego.model.exception import MaltegoException


@register_transform(...)
async def validate_input(entity: Phrase, context: MaltegoContext):
    if not entity.value.isnumeric():
        raise MaltegoException(
            "Invalid input: Expected a numeric value. "
            f"Got '{entity.value}' instead."
        )
    return Phrase(f"Valid: {entity.value}")

The user sees: "Invalid input: Expected a numeric value. Got 'abc' instead."

MaltegoTransformTimeoutError

For timeout-specific failures:

from maltego.model.exception import MaltegoTransformTimeoutError


@register_transform(...)
async def slow_operation(entity: Phrase, context: MaltegoContext):
    try:
        result = await asyncio.wait_for(fetch_data(), timeout=30)
    except asyncio.TimeoutError:
        raise MaltegoTransformTimeoutError(
            "Operation timed out after 30 seconds. Try a smaller query."
        )
    return result

IntegrationClient Exceptions

IntegrationClient raises specific exceptions for different error scenarios:

from maltego.model.exception import (
    MaltegoException,
    MaltegoHTTPDataProviderAPIKeyInvalid,  # 401 Unauthorized
    MaltegoHTTPUnauthorized,               # 403 Forbidden
    MaltegoHTTPDataProviderNotFound,       # 404 Not Found
    MaltegoHTTPDataProviderUnavailable,    # 5xx, timeouts, connection errors
)


try:
    response = await client.get(url, context)
except MaltegoHTTPDataProviderNotFound:
    context.log.inform("Resource not found")
    return None
except MaltegoHTTPDataProviderUnavailable as e:
    context.log.fatal(f"Service unavailable: {e.message}")
    return None
except MaltegoException as e:
    # Catch-all for other errors (400, 429, etc.)
    context.log.fatal(f"Error: {e.message}")
    return None

Unhandled Exceptions

Unhandled exceptions (like ``ValueError``, ``KeyError``, etc.) show a **generic error message** to users. The actual error is logged server-side but hidden from users for security.

# BAD: User sees generic transform failed message
@register_transform(...)
async def bad_error_handling(entity: Phrase, context: MaltegoContext):
    raise ValueError("User won't see this message")

# GOOD: User sees your specific message
@register_transform(...)
async def good_error_handling(entity: Phrase, context: MaltegoContext):
    raise MaltegoException("Something went wrong. Please try again.")

Error Handling Patterns

Pattern 1: Validate and Fail Early

@register_transform(...)
async def validated_transform(entity: Phrase, context: MaltegoContext):
    # Validate input first
    if not entity.value:
        raise MaltegoException("Input cannot be empty")

    if len(entity.value) > 100:
        raise MaltegoException("Input too long (max 100 characters)")

    # Proceed with valid input
    return Phrase(f"Valid: {entity.value}")

Pattern 2: Graceful API Error Handling

from maltego.server import IntegrationClient
from maltego.model.exception import (
    MaltegoException,
    MaltegoHTTPDataProviderNotFound,
    MaltegoHTTPDataProviderUnavailable,
)

client = IntegrationClient()

@register_transform(...)
async def api_transform(entity: Phrase, context: MaltegoContext):
    try:
        response = await client.get(f"https://api.example.com/{entity.value}", context)
        return Phrase(response.json()["result"])

    except MaltegoHTTPDataProviderNotFound:
        context.log.inform("Resource not found")
        return None

    except MaltegoHTTPDataProviderUnavailable as e:
        context.log.fatal(f"Service unavailable: {e.message}")
        return None

    except MaltegoException as e:
        # Catch-all for other errors (400, 429, connection failures, etc.)
        context.log.fatal(f"Error: {e.message}")
        return None

Pattern 3: Partial Results on Failure

Return what you can when some operations fail:

@register_transform(...)
async def batch_with_errors(entity: Phrase, context: MaltegoContext):
    items = ["a", "b", "c", "d"]
    results = []
    errors = []


    for item in items:
        try:
            result = await process_item(item)
            results.append(Phrase(result))
        except Exception as e:
            errors.append(item)
            context.log.debug(f"Failed {item}: {e}")


    if errors:
        context.log.fatal(f"Failed to process: {', '.join(errors)}")


    context.log.inform(f"Completed: {len(results)} success, {len(errors)} failed")
    return results  # Return partial results

Pattern 4: Retry Logic

import asyncio

async def fetch_with_retry(url: str, context: MaltegoContext, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return await client.get(url, context)
        except MaltegoException as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                context.log.partial(f"Retry {attempt + 1}/{max_retries} in {wait_time}s...")
                await asyncio.sleep(wait_time)
            else:
                raise MaltegoException(f"Failed after {max_retries} attempts: {e.message}")

Complete Example

Run maltego-transforms start my_project to create a new project from the template. See transforms/error_handling_example.py for runnable examples of all error handling patterns covered in this guide.

Best Practices

Use MaltegoException for user-facing errors - Always provide helpful messages
Catch specific exceptions - Handle different error types appropriately
Log errors before handling - Use context.log.fatal() for visibility
Return partial results - Don't fail entirely if some items succeed
Validate input early - Fail fast with clear messages
Hide internal details - Don't expose stack traces or sensitive info to users