We Switched from Contractors to Full-Time Engineers: Saved 40% with Deel and Rippling

import os
import time
import logging
from typing import List, Dict, Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("deel_rippling_sync.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Constants for API endpoints and auth
DEEL_API_BASE = "https://api.deel.com/v1"
RIPPLING_API_BASE = "https://api.rippling.com/v3"
MAX_RETRIES = 3
RETRY_BACKOFF = 2

class DeelRipplingSyncer:
    def __init__(self, deel_api_key: str, rippling_api_key: str):
        self.deel_session = self._init_session(deel_api_key, "Deel")
        self.rippling_session = self._init_session(rippling_api_key, "Rippling")
        self.synced_contractors = 0
        self.failed_syncs = 0

    def _init_session(self, api_key: str, service: str) -> requests.Session:
        """Initialize a requests session with retry logic and auth headers."""
        session = requests.Session()
        retry_strategy = Retry(
            total=MAX_RETRIES,
            backoff_factor=RETRY_BACKOFF,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST", "PUT"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "User-Agent": "DeelRipplingSyncer/1.0 (senior-eng-blog)"
        })
        logger.info(f"Initialized {service} session with retry logic")
        return session

    def fetch_deel_contractors(self, team_id: str) -> List[Dict]:
        """Fetch all active contractors from Deel for a given team."""
        contractors = []
        page = 1
        while True:
            try:
                response = self.deel_session.get(
                    f"{DEEL_API_BASE}/contractors",
                    params={"team_id": team_id, "status": "active", "page": page, "per_page": 100}
                )
                response.raise_for_status()
                data = response.json()
                if not data.get("contractors"):
                    break
                contractors.extend(data["contractors"])
                logger.info(f"Fetched page {page} of Deel contractors: {len(data['contractors'])} records")
                page += 1
                time.sleep(0.5)  # Respect rate limits
            except requests.exceptions.RequestException as e:
                logger.error(f"Failed to fetch Deel contractors page {page}: {str(e)}")
                raise
        logger.info(f"Total active Deel contractors fetched: {len(contractors)}")
        return contractors

    def map_to_rippling_employee(self, contractor: Dict) -> Optional[Dict]:
        """Map Deel contractor fields to Rippling employee schema."""
        try:
            return {
                "external_id": contractor["id"],
                "first_name": contractor["first_name"],
                "last_name": contractor["last_name"],
                "email": contractor["email"],
                "job_title": contractor["role"],
                "employment_type": "full_time",  # We're converting contractors to full-time
                "start_date": contractor["contract_start_date"],
                "salary": {
                    "amount": contractor["hourly_rate"] * 160,  # Assume 160 hours/month for full-time
                    "currency": contractor["currency"],
                    "payment_frequency": "monthly"
                },
                "department": contractor["team"],
                "location": {
                    "country": contractor["country"],
                    "state": contractor.get("state", ""),
                    "city": contractor.get("city", "")
                }
            }
        except KeyError as e:
            logger.error(f"Missing required field {str(e)} for contractor {contractor.get('id')}")
            return None

    def sync_to_rippling(self, rippling_employee: Dict) -> bool:
        """Create or update an employee in Rippling."""
        try:
            # Check if employee already exists via external_id
            existing = self.rippling_session.get(
                f"{RIPPLING_API_BASE}/employees",
                params={"external_id": rippling_employee["external_id"]}
            ).json()
            if existing.get("data"):
                # Update existing employee
                emp_id = existing["data"][0]["id"]
                response = self.rippling_session.put(
                    f"{RIPPLING_API_BASE}/employees/{emp_id}",
                    json=rippling_employee
                )
                logger.info(f"Updated existing Rippling employee {emp_id}")
            else:
                # Create new employee
                response = self.rippling_session.post(
                    f"{RIPPLING_API_BASE}/employees",
                    json=rippling_employee
                )
                logger.info(f"Created new Rippling employee {response.json()['id']}")
            response.raise_for_status()
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to sync employee {rippling_employee.get('external_id')}: {str(e)}")
            return False

    def run_sync(self, team_id: str):
        """Orchestrate the full sync process."""
        logger.info(f"Starting sync for Deel team {team_id}")
        contractors = self.fetch_deel_contractors(team_id)
        for contractor in contractors:
            mapped = self.map_to_rippling_employee(contractor)
            if not mapped:
                self.failed_syncs += 1
                continue
            if self.sync_to_rippling(mapped):
                self.synced_contractors += 1
            else:
                self.failed_syncs += 1
        logger.info(f"Sync complete. Synced: {self.synced_contractors}, Failed: {self.failed_syncs}")

if __name__ == "__main__":
    # Load API keys from environment variables (never hardcode!)
    deel_key = os.getenv("DEEL_API_KEY")
    rippling_key = os.getenv("RIPPLING_API_KEY")
    team_id = os.getenv("DEEL_TEAM_ID")

    if not all([deel_key, rippling_key, team_id]):
        logger.error("Missing required environment variables: DEEL_API_KEY, RIPPLING_API_KEY, DEEL_TEAM_ID")
        exit(1)

    syncer = DeelRipplingSyncer(deel_key, rippling_key)
    syncer.run_sync(team_id)

import csv
import json
from typing import List, Dict, Tuple
from dataclasses import dataclass
import logging

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

@dataclass
class ContractorCost:
    id: str
    hourly_rate: float
    hours_per_month: int
    benefits_markup: float  # 0.0 for contractors, ~30% for full-time
    overhead_per_month: float  # Compliance, tools, etc.

@dataclass
class FullTimeCost:
    id: str
    annual_salary: float
    equity_value: float  # Monthly vesting value
    benefits_rate: float  # 0.3 for 30% benefits
    overhead_per_month: float

class CostComparisonCalculator:
    def __init__(self, contractors: List[ContractorCost], full_time: List[FullTimeCost]):
        self.contractors = contractors
        self.full_time = full_time
        self.total_contractor_cost = 0.0
        self.total_full_time_cost = 0.0

    def calculate_contractor_tco(self) -> float:
        """Calculate total cost of ownership for contractors over 12 months."""
        total = 0.0
        for c in self.contractors:
            try:
                monthly_cost = (c.hourly_rate * c.hours_per_month) * (1 + c.benefits_markup) + c.overhead_per_month
                annual_cost = monthly_cost * 12
                total += annual_cost
                logger.debug(f"Contractor {c.id} annual cost: ${annual_cost:,.2f}")
            except Exception as e:
                logger.error(f"Failed to calculate cost for contractor {c.id}: {str(e)}")
                raise
        self.total_contractor_cost = total
        logger.info(f"Total contractor TCO (12mo): ${total:,.2f}")
        return total

    def calculate_full_time_tco(self) -> float:
        """Calculate total cost of ownership for full-time engineers over 12 months."""
        total = 0.0
        for ft in self.full_time:
            try:
                annual_benefits = ft.annual_salary * ft.benefits_rate
                annual_equity = ft.equity_value * 12
                annual_overhead = ft.overhead_per_month * 12
                annual_cost = ft.annual_salary + annual_benefits + annual_equity + annual_overhead
                total += annual_cost
                logger.debug(f"Full-time {ft.id} annual cost: ${annual_cost:,.2f}")
            except Exception as e:
                logger.error(f"Failed to calculate cost for full-time {ft.id}: {str(e)}")
                raise
        self.total_full_time_cost = total
        logger.info(f"Total full-time TCO (12mo): ${total:,.2f}")
        return total

    def calculate_savings(self) -> Tuple[float, float]:
        """Return absolute savings and percentage savings."""
        if self.total_contractor_cost == 0:
            raise ValueError("Contractor TCO not calculated yet")
        savings = self.total_contractor_cost - self.total_full_time_cost
        percentage = (savings / self.total_contractor_cost) * 100
        logger.info(f"Total savings: ${savings:,.2f} ({percentage:.1f}%)")
        return savings, percentage

    def export_to_csv(self, filename: str):
        """Export cost breakdown to CSV for audit."""
        try:
            with open(filename, "w", newline="") as f:
                writer = csv.writer(f)
                writer.writerow(["Type", "ID", "Annual Cost"])
                for c in self.contractors:
                    monthly = (c.hourly_rate * c.hours_per_month) * (1 + c.benefits_markup) + c.overhead_per_month
                    writer.writerow(["Contractor", c.id, monthly * 12])
                for ft in self.full_time:
                    annual = ft.annual_salary * (1 + ft.benefits_rate) + (ft.equity_value *12) + (ft.overhead_per_month *12)
                    writer.writerow(["Full-Time", ft.id, annual])
            logger.info(f"Exported cost breakdown to {filename}")
        except IOError as e:
            logger.error(f"Failed to export CSV: {str(e)}")
            raise

if __name__ == "__main__":
    # Sample data from our 2023-2024 switch (anonymized)
    contractors = [
        ContractorCost("c_001", 185.0, 120, 0.0, 450.0),  # Senior contractor, 120h/mo
        ContractorCost("c_002", 145.0, 160, 0.0, 450.0),  # Mid contractor, 160h/mo
        ContractorCost("c_003", 185.0, 100, 0.0, 450.0),  # Senior, part-time
        # ... 19 more contractors (total 22)
    ]

    full_time = [
        FullTimeCost("ft_001", 185000.0, 2500.0, 0.3, 280.0),  # Senior FT, $185k salary
        FullTimeCost("ft_002", 145000.0, 1800.0, 0.3, 280.0),  # Mid FT, $145k salary
        FullTimeCost("ft_003", 185000.0, 2500.0, 0.3, 280.0),  # Senior FT
        # ... 15 more full-time (total 18)
    ]

    calculator = CostComparisonCalculator(contractors, full_time)
    calculator.calculate_contractor_tco()
    calculator.calculate_full_time_tco()
    savings, pct = calculator.calculate_savings()
    print(f"💰 Total annual savings: ${savings:,.2f} ({pct:.1f}%)")
    calculator.export_to_csv("cost_comparison.csv")

import hmac
import hashlib
import json
from flask import Flask, request, jsonify
from typing import Dict, List
import logging
import time

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

app = Flask(__name__)

# Webhook secrets from Deel and Rippling (store in env vars)
DEEL_WEBHOOK_SECRET = "deel_webhook_secret_placeholder"
RIPPLING_WEBHOOK_SECRET = "rippling_webhook_secret_placeholder"

# In-memory store for onboarding events (use Redis in prod)
onboarding_events: Dict[str, List[Dict]] = {}

class OnboardingTracker:
    @staticmethod
    def verify_deel_signature(payload: bytes, signature: str) -> bool:
        """Verify Deel webhook signature using HMAC-SHA256."""
        try:
            expected = hmac.new(
                DEEL_WEBHOOK_SECRET.encode(),
                payload,
                hashlib.sha256
            ).hexdigest()
            return hmac.compare_digest(expected, signature)
        except Exception as e:
            logger.error(f"Deel signature verification failed: {str(e)}")
            return False

    @staticmethod
    def verify_rippling_signature(payload: bytes, signature: str) -> bool:
        """Verify Rippling webhook signature using HMAC-SHA256."""
        try:
            expected = hmac.new(
                RIPPLING_WEBHOOK_SECRET.encode(),
                payload,
                hashlib.sha256
            ).hexdigest()
            return hmac.compare_digest(expected, signature)
        except Exception as e:
            logger.error(f"Rippling signature verification failed: {str(e)}")
            return False

    @staticmethod
    def track_event(employee_id: str, event_type: str, timestamp: float, metadata: Dict):
        """Track an onboarding event for an employee."""
        if employee_id not in onboarding_events:
            onboarding_events[employee_id] = []
        event = {
            "type": event_type,
            "timestamp": timestamp,
            "metadata": metadata
        }
        onboarding_events[employee_id].append(event)
        logger.info(f"Tracked {event_type} for {employee_id} at {timestamp}")

    @staticmethod
    def calculate_onboarding_time(employee_id: str) -> float:
        """Calculate total onboarding time in hours between first and last event."""
        events = onboarding_events.get(employee_id, [])
        if not events:
            return 0.0
        # Sort events by timestamp
        sorted_events = sorted(events, key=lambda x: x["timestamp"])
        start = sorted_events[0]["timestamp"]
        end = sorted_events[-1]["timestamp"]
        return (end - start) / 3600  # Convert seconds to hours

@app.route("/webhooks/deel", methods=["POST"])
def deel_webhook():
    """Handle Deel contractor onboarding webhooks."""
    payload = request.data
    signature = request.headers.get("X-Deel-Signature", "")

    if not OnboardingTracker.verify_deel_signature(payload, signature):
        logger.warning("Invalid Deel webhook signature")
        return jsonify({"error": "Invalid signature"}), 401

    data = json.loads(payload)
    employee_id = data.get("contractor_id")
    event_type = data.get("event_type")
    timestamp = data.get("timestamp", time.time())

    if event_type == "contractor.hired":
        OnboardingTracker.track_event(employee_id, "deel_hire", timestamp, data)
    elif event_type == "contractor.onboarded":
        OnboardingTracker.track_event(employee_id, "deel_onboarded", timestamp, data)

    return jsonify({"status": "ok"}), 200

@app.route("/webhooks/rippling", methods=["POST"])
def rippling_webhook():
    """Handle Rippling employee onboarding webhooks."""
    payload = request.data
    signature = request.headers.get("X-Rippling-Signature", "")

    if not OnboardingTracker.verify_rippling_signature(payload, signature):
        logger.warning("Invalid Rippling webhook signature")
        return jsonify({"error": "Invalid signature"}), 401

    data = json.loads(payload)
    employee_id = data.get("employee_id")
    event_type = data.get("event_type")
    timestamp = data.get("timestamp", time.time())

    if event_type == "employee.created":
        OnboardingTracker.track_event(employee_id, "rippling_created", timestamp, data)
    elif event_type == "employee.onboarded":
        OnboardingTracker.track_event(employee_id, "rippling_onboarded", timestamp, data)
        # Calculate total onboarding time
        total_hours = OnboardingTracker.calculate_onboarding_time(employee_id)
        logger.info(f"Total onboarding time for {employee_id}: {total_hours:.1f} hours")

    return jsonify({"status": "ok"}), 200

@app.route("/metrics/onboarding", methods=["GET"])
def get_onboarding_metrics():
    """Return onboarding time metrics for all employees."""
    metrics = []
    for emp_id, events in onboarding_events.items():
        total_hours = OnboardingTracker.calculate_onboarding_time(emp_id)
        metrics.append({
            "employee_id": emp_id,
            "event_count": len(events),
            "total_onboarding_hours": round(total_hours, 2)
        })
    return jsonify(metrics), 200

if __name__ == "__main__":
    logger.info("Starting onboarding tracker on port 5000")
    app.run(host="0.0.0.0", port=5000, debug=False)

def validate_contractor_compliance(self, contractor_id: str) -> bool:
    """Validate contractor compliance via Deel API before sync."""
    try:
        response = self.deel_session.get(
            f"{DEEL_API_BASE}/compliance/validate",
            params={"contractor_id": contractor_id}
        )
        response.raise_for_status()
        result = response.json()
        if not result.get("is_compliant"):
            logger.error(f"Contractor {contractor_id} non-compliant: {result.get('violations')}")
            return False
        logger.info(f"Contractor {contractor_id} compliance validated")
        return True
    except requests.exceptions.RequestException as e:
        logger.error(f"Compliance check failed for {contractor_id}: {str(e)}")
        return False

# Add to the Rippling webhook handler to track equipment assignment
if event_type == "equipment.assigned":
    OnboardingTracker.track_event(employee_id, "equipment_assigned", timestamp, {
        "equipment_id": data.get("equipment_id"),
        "type": data.get("type")  # Laptop, monitor, etc.
    })

# Extend CostComparisonCalculator to pull real-time contractor rates
def fetch_deel_rates(self, contractor_id: str) -> float:
    """Fetch current hourly rate for a contractor from Deel API."""
    response = self.deel_session.get(f"{DEEL_API_BASE}/contractors/{contractor_id}/rates")
    response.raise_for_status()
    return response.json()["current_rate"]