Callbacks and Hooks

Callbacks and Hooks

Fiscus SDK provides a robust and flexible system for callbacks and hooks to customize task flows, handle responses, and integrate custom logic into your API operations. This section dives deep into the available callbacks and hooks that allow you to personalize your workflows for more advanced use cases.

Callbacks

Callbacks in Fiscus SDK enable you to handle specific events during the execution of API tasks. They allow you to manage success, error, authentication, streaming, and logging scenarios dynamically as the task progresses.

Success Callbacks

Success callbacks are triggered when an API task completes successfully. You can use this callback to:

• Log successful operations
• Trigger downstream workflows
• Update your application's state

Python

Example: Success Callback

1. Define a Success Callback
   Define a function that takes an info dictionary with details about the successful task.

def fiscus_on_success(info):
    print(f"Operation succeeded: {info['message']}")

2. Attach the Callback to Your Task
   Add the callback when executing the API task by passing it in the callbacks parameter.

callbacks = {'fiscus_on_success': fiscus_on_success}

response = client.execute('Gmail', 'send_email', {
    'to': 'recipient@example.com',
    'subject': 'Test Email',
    'body': 'Hello!'
}, callbacks=callbacks, user=user)

3. Output on Success
   When the task is successful, the fiscus_on_success callback will print out a custom success message.

Error Callbacks

Error callbacks are essential for handling failure scenarios. You can log errors, retry failed tasks, or even notify your team if something goes wrong.

Example: Error Callback

1. Define an Error Callback
   The info dictionary contains detailed information about the error, including the reason for failure.

def fiscus_on_error(info):
    print(f"Error occurred: {info['message']}")

2. Attach the Callback to Your Task
   Include the callback in the callbacks parameter when executing the task.

callbacks = {'fiscus_on_error': fiscus_on_error}

response = client.execute('Slack', 'post_message', {
    'channel': 'general',
    'text': 'Hello, Slack!'
}, callbacks=callbacks, user=user)

3. Output on Error
   If the task fails, the fiscus_on_error callback will output the error message, helping you diagnose the issue quickly.

Authentication Callbacks

For services that require dynamic authentication, the authentication callback is your go-to. This allows you to handle API key refreshes, OAuth tokens, or other authentication flows specific to the service you're connecting to.

Example: Authentication Callback

1. Define an Authentication Callback
   Depending on the service, return the required authentication credentials.

def fiscus_on_auth(connector_name):
    if connector_name == 'QuickBooks':
        return {
            'client_id': 'YOUR_CLIENT_ID',
            'client_secret': 'YOUR_CLIENT_SECRET',
            'access_token': 'USER_ACCESS_TOKEN',
            'refresh_token': 'USER_REFRESH_TOKEN'
        }
    elif connector_name == 'Salesforce':
        return {
            'username': 'YOUR_USERNAME',
            'password': 'YOUR_PASSWORD',
            'token': 'YOUR_SECURITY_TOKEN'
        }
    else:
        return None

2. Set the Authentication Callback
   Attach the authentication callback to the user.

user.set_auth_callback(fiscus_on_auth)

3. Auto-Authentication
   The callback automatically retrieves the correct credentials when an API call is made, keeping your workflows secure and seamless.

Streaming Callbacks

For real-time data or long-running operations, streaming callbacks are used to handle data that comes in asynchronously. This is ideal for WebSocket connections or live data feeds.

Example: Streaming Callback

1. Define a Streaming Callback
   Process each piece of data as it comes in through the stream.

def fiscus_on_stream(data):
    print(f"Streaming data received: {data}")

2. Attach the Streaming Callback
   Include the callback in your task execution.

callbacks = {'fiscus_on_stream': fiscus_on_stream}

response = client.execute('StockAPI', 'subscribe_ticker', {
    'ticker': 'AAPL'
}, callbacks=callbacks, user=user)

3. Real-Time Data Handling
   Every time new data is received from the stream, it’s processed by the fiscus_on_stream callback.

Logging and Debugging Callbacks

Logging and debugging callbacks are perfect for monitoring the execution of tasks. Use them to log important steps, track progress, and debug complex workflows in real-time.

Example: Logging and Debugging Callback

1. Define a Logging Callback
   Capture logs with detailed messages about the task's progress.

def fiscus_on_log(info):
    print(f"Log: {info['message']}")

2. Attach the Logging Callback
   Attach the callback when executing your task.

callbacks = {'fiscus_on_log': fiscus_on_log}

response = client.execute('SomeService', 'some_operation', {...}, callbacks=callbacks, user=user)

3. Monitor Execution
   As the task executes, each step will be logged to help you track its progress.

 

Customization Hooks

While callbacks handle specific task events, hooks allow you to customize the behavior before and after every task is executed. They give you the flexibility to inject logic at critical points in the workflow, enabling you to modify task parameters, log key data, or respond to results dynamically.

Pre-Execution and Post-Execution Hooks

Hooks are defined globally for a client and apply to all tasks executed by that client. This is perfect for cross-cutting concerns like logging, monitoring, or modifying input parameters globally.

Pre-Execution Hooks

Pre-execution hooks run before an API call is executed, allowing you to adjust the request parameters, log key data, or even conditionally block a request from being sent.

Use Cases for Pre-Execution Hooks:

• Adding timestamps to API calls
• Injecting authentication tokens
• Logging the parameters of every task before execution

Example: Pre-Execution Hook

1. Define a Pre-Execution Hook

def pre_execution_hook(connector_name, operation_name, params):
    print(f"About to execute {operation_name} on {connector_name} with params: {params}")
    params['timestamp'] = '2024-01-01T12:00:00Z'  # Add a timestamp
    return params

2. Attach the Hook

   Use set_pre_execution_hook to apply it to all API tasks for this client.

client.set_pre_execution_hook(pre_execution_hook)

3. Make the API Call

   The pre-execution hook will run before every task, adjusting parameters as needed.

Post-Execution Hooks

Post-execution hooks are invoked after the API call has been completed. They allow you to inspect the response, log results, or trigger follow-up tasks.

Use Cases for Post-Execution Hooks:

• Logging success or failure
• Triggering follow-up tasks
• Modifying the response before passing it to the next step

Example: Post-Execution Hook

1. Define a Post-Execution Hook

def post_execution_hook(response):
    if response.success:
        print(f"Task succeeded: {response.result}")
    else:
        print(f"Task failed: {response.error_message}")
    return response

2. Attach the Hook

   Use set_post_execution_hook to apply it globally.

client.set_post_execution_hook(post_execution_hook)

3. Inspect the Results

   After every task execution, the post-execution hook will inspect and log the results.

Customizing Task Flows with Hooks

The real power of hooks comes when you combine them with callbacks to build complex, multi-step workflows. By dynamically modifying task inputs, handling success and error cases, and chaining multiple tasks together, you can create highly customized API orchestrations.

Example: Dynamic Task Flow Customization

Consider a flow where you:

1. Create a new customer in a CRM system.

2. Send a welcome email after the customer is created.

3. Log the workflow and mark it as complete.

4. Define Hooks and Callbacks

• Pre-Execution Hook: Injects a timestamp for record creation.

def pre_execution_hook(connector_name, operation_name, params):
    if operation_name == 'create_customer':
        params['created_at'] = '2024-01-01T12:00:00Z'
    return params

• Post-Execution Hook: Automatically triggers the next task after a successful customer creation.

def post_execution_hook(response):
    if response.success and response.operation_name == 'create_customer':
        client.execute('EmailService', 'send_welcome_email', {'to': response.result['email']})
    return response

• Success Callback: Logs the entire workflow once the email is sent.

def on_success(info):
    print(f"Workflow completed successfully: {info['message']}")

• Error Callback: Handles any failure by logging and retrying.

def on_error(info):
    print(f"Workflow failed: {info['message']} - Retrying...")
    retry_params = {...}
    client.execute('CRMSystem', 'create_customer', retry_params)

2. Attach the Hooks Globally

client.set_pre_execution_hook(pre_execution_hook)
client.set_post_execution_hook(post_execution_hook)

3. Execute the Workflow

callbacks = {'fiscus_on_success': on_success, 'fiscus_on_error': on_error}
client.execute('CRMSystem', 'create_customer', customer_params, callbacks=callbacks)

By utilizing both hooks and callbacks, you unlock the full potential of the Fiscus SDK for building dynamic, resilient, and highly customizable API workflows. Whether you're chaining complex multi-step tasks or simply adding logging for easier debugging, these tools give you complete control over the task lifecycle.