Best n8n Alternatives for Enterprise API Automation

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.postgres_operator import PostgresOperator
from airflow.models import Variable
import requests
import json
import psycopg2
from psycopg2.extras import execute_batch

default_args = {
    'owner': 'data_team',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
    'email': ['alerts@company.com'],
    'start_date': datetime(2024, 1, 1),
}

dag = DAG(
    'stripe_to_postgres_sync',
    default_args=default_args,
    description='Sync Stripe customers to Postgres hourly',
    schedule_interval='0 * * * *',
    catchup=False,
    tags=['data-pipeline', 'stripe', 'critical'],
)

def fetch_stripe_customers(**context):
    stripe_api_key = Variable.get('stripe_api_key')
    url = 'https://api.stripe.com/v1/customers'
    headers = {'Authorization': f'Bearer {stripe_api_key}'}
    
    all_customers = []
    has_more = True
    starting_after = None
    
    while has_more:
        params = {'limit': 100}
        if starting_after:
            params['starting_after'] = starting_after
        
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        data = response.json()
        
        all_customers.extend(data['data'])
        has_more = data['has_more']
        
        if all_customers:
            starting_after = all_customers[-1]['id']
    
    context['task_instance'].xcom_push(key='customers', value=all_customers)
    return f"Fetched {len(all_customers)} customers"

def transform_customer_data(**context):
    customers = context['task_instance'].xcom_pull(task_ids='fetch_stripe_customers', key='customers')
    
    transformed = []
    for customer in customers:
        transformed.append({
            'stripe_id': customer['id'],
            'email': customer.get('email'),
            'name': customer.get('name'),
            'created_at': datetime.fromtimestamp(customer['created']),
            'balance': customer.get('balance', 0),
            'updated_at': datetime.now(),
        })
    
    context['task_instance'].xcom_push(key='transformed_customers', value=transformed)
    return f"Transformed {len(transformed)} records"

def load_to_postgres(**context):
    transformed = context['task_instance'].xcom_pull(task_ids='transform_customer_data', key='transformed_customers')
    
    db_conn = psycopg2.connect(
        host=Variable.get('postgres_host'),
        database=Variable.get('postgres_db'),
        user=Variable.get('postgres_user'),
        password=Variable.get('postgres_password'),
    )
    
    cursor = db_conn.cursor()
    
    insert_query = """
    INSERT INTO stripe_customers (stripe_id, email, name, created_at, balance, updated_at)
    VALUES (%s, %s, %s, %s, %s, %s)
    ON CONFLICT (stripe_id) DO UPDATE SET
        email = EXCLUDED.email,
        name = EXCLUDED.name,
        balance = EXCLUDED.balance,
        updated_at = EXCLUDED.updated_at
    """
    
    execute_batch(cursor, insert_query, [
        (c['stripe_id'], c['email'], c['name'], c['created_at'], c['balance'], c['updated_at'])
        for c in transformed
    ], page_size=1000)
    
    db_conn.commit()
    cursor.close()
    db_conn.close()
    
    return f"Loaded {len(transformed)} records to Postgres"

fetch_task = PythonOperator(
    task_id='fetch_stripe_customers',
    python_callable=fetch_stripe_customers,
    provide_context=True,
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform_customer_data',
    python_callable=transform_customer_data,
    provide_context=True,
    dag=dag,
)

load_task = PythonOperator(
    task_id='load_to_postgres',
    python_callable=load_to_postgres,
    provide_context=True,
    dag=dag,
)

fetch_task >> transform_task >> load_task

import * as wf from '@temporalio/workflow';
import { proxyActivities } from '@temporalio/workflow';
import type { Activities } from './activities';

const { fetchPaymentFromStripe, fetchOrderFromDatabase, comparePayments, notifyFinanceTeam, retryFailedPayment } = proxyActivities<Activities>({
  startToCloseTimeout: '5 minutes',
  retry: {
    initialInterval: '1 second',
    maximumInterval: '1 minute',
    maximumAttempts: 3,
  },
});

export interface PaymentReconciliationInput {
  orderId: string;
  paymentId: string;
  expectedAmount: number;
}

export async function paymentReconciliationWorkflow(input: PaymentReconciliationInput): Promise<string> {
  const { orderId, paymentId, expectedAmount } = input;

  try {
    // Fetch data in parallel
    const [stripePayment, dbOrder] = await Promise.all([
      fetchPaymentFromStripe(paymentId),
      fetchOrderFromDatabase(orderId),
    ]);

    // Compare amounts
    const discrepancy = await comparePayments(stripePayment.amount, expectedAmount);

    if (discrepancy > 0) {
      // Wait for human approval before retrying
      await wf.waitForSignal<boolean>('approveRetry');
      await retryFailedPayment(paymentId, expectedAmount);
      return `Payment corrected after approval`;
    }

    if (discrepancy < 0) {
      // Refund difference
      await notifyFinanceTeam(`Overpayment detected: ${Math.abs(discrepancy)} cents`, orderId);
      return `Finance team notified of overpayment`;
    }

    return `Payment reconciled successfully`;
  } catch (error) {
    await notifyFinanceTeam(`Reconciliation failed: ${error.message}`, orderId);
    throw error;
  }
}

import Stripe from 'stripe';
import postgres from 'pg';

const stripe = new Stripe(process.env.STRIPE_API_KEY);
const pool = new postgres.Pool({
  connectionString: process.env.DATABASE_URL,
});

export const activities = {
  async fetchPaymentFromStripe(paymentId: string) {
    const charge = await stripe.charges.retrieve(paymentId);
    return {
      id: charge.id,
      amount: charge.amount,
      currency: charge.currency,
      status: charge.status,
      created: charge.created,
    };
  },

  async fetchOrderFromDatabase(orderId: string) {
    const client = await pool.connect();
    try {
      const result = await client.query(
        'SELECT id, user_id, total_amount, status, created_at FROM orders WHERE id = $1',
        [orderId]