Step-by-Step Guide to Setting Up Time Tracking for 100+ Developers Using Toggl 5.0, Slack 5.0, and BambooHR 5.0

import flask
import hmac
import hashlib
import redis
import json
import os
import logging
from datetime import datetime, timezone
import traceback

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

app = flask.Flask(__name__)

# Load config from environment variables (never hardcode secrets!)
BAMBOO_WEBHOOK_SECRET = os.getenv("BAMBOO_WEBHOOK_SECRET")
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)

# Initialize Redis client with connection pooling
try:
    redis_client = redis.Redis(
        host=REDIS_HOST,
        port=REDIS_PORT,
        password=REDIS_PASSWORD,
        decode_responses=True,
        socket_connect_timeout=5,
        retry_on_timeout=True
    )
    # Test Redis connection on startup
    redis_client.ping()
    logger.info("Redis connection established successfully")
except redis.ConnectionError as e:
    logger.error(f"Failed to connect to Redis: {str(e)}")
    raise SystemExit(1)

def validate_bamboo_signature(payload_body: bytes, signature_header: str) -> bool:
    """Validate BambooHR 5.0 webhook signature using HMAC-SHA256.
    BambooHR sends X-Bamboo-Signature header with HMAC of payload + secret.
    """
    if not BAMBOO_WEBHOOK_SECRET:
        logger.warning("No BAMBOO_WEBHOOK_SECRET set, skipping signature validation")
        return True
    if not signature_header:
        logger.error("Missing X-Bamboo-Signature header")
        return False
    # Compute HMAC of raw payload
    digest = hmac.new(
        BAMBOO_WEBHOOK_SECRET.encode("utf-8"),
        payload_body,
        hashlib.sha256
    ).hexdigest()
    # Compare using constant time comparison to prevent timing attacks
    return hmac.compare_digest(digest, signature_header)

@app.route("/bamboo-webhook", methods=["POST"])
def handle_bamboo_webhook():
    """Handle BambooHR 5.0 employee change webhooks.
    Supported events: employee.hire, employee.termination, employee.role_change
    """
    # Get raw payload for signature validation
    payload_body = flask.request.get_data()
    signature = flask.request.headers.get("X-Bamboo-Signature")

    # Validate webhook signature
    if not validate_bamboo_signature(payload_body, signature):
        logger.error("Invalid BambooHR webhook signature")
        return flask.jsonify({"error": "Invalid signature"}), 401

    # Parse payload
    try:
        payload = json.loads(payload_body)
    except json.JSONDecodeError as e:
        logger.error(f"Invalid JSON payload: {str(e)}")
        return flask.jsonify({"error": "Invalid JSON"}), 400

    event_type = payload.get("eventType")
    if not event_type:
        logger.error("Missing eventType in payload")
        return flask.jsonify({"error": "Missing eventType"}), 400

    # Process supported events only
    supported_events = ["employee.hire", "employee.termination", "employee.role_change"]
    if event_type not in supported_events:
        logger.info(f"Ignoring unsupported event: {event_type}")
        return flask.jsonify({"status": "ignored"}), 200

    employee_data = payload.get("employee")
    if not employee_data:
        logger.error("Missing employee data in payload")
        return flask.jsonify({"error": "Missing employee data"}), 400

    employee_id = employee_data.get("id")
    if not employee_id:
        logger.error("Missing employee ID in payload")
        return flask.jsonify({"error": "Missing employee ID"}), 400

    # Extract billable rate, role, employment status
    try:
        billable_rate = float(employee_data.get("billableRate", 0.0))
        role = employee_data.get("jobTitle", "Unknown")
        is_active = not (event_type == "employee.termination")
        # Cache key: bamboo:employee:{id}
        cache_key = f"bamboo:employee:{employee_id}"
        cache_data = {
            "billable_rate": billable_rate,
            "role": role,
            "is_active": is_active,
            "last_updated": datetime.now(timezone.utc).isoformat()
        }
        # Set cache with 24 hour TTL (BambooHR data changes infrequently)
        redis_client.setex(cache_key, 86400, json.dumps(cache_data))
        logger.info(f"Updated cache for employee {employee_id}: {cache_data}")
    except (ValueError, TypeError) as e:
        logger.error(f"Failed to parse employee data for {employee_id}: {str(e)}")
        return flask.jsonify({"error": "Invalid employee data"}), 400
    except redis.RedisError as e:
        logger.error(f"Redis error updating employee {employee_id}: {str(e)}")
        return flask.jsonify({"error": "Cache update failed"}), 500

    return flask.jsonify({"status": "success"}), 200

if __name__ == "__main__":
    # Run with production WSGI server in real deployments (gunicorn)
    app.run(host="0.0.0.0", port=8080, debug=False)

import requests
import redis
import psycopg2
import json
import os
import logging
from datetime import datetime, timedelta, timezone
from typing import List, Dict
import time

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

# Load config from env
TOGGL_API_KEY = os.getenv("TOGGL_API_KEY")
TOGGL_WORKSPACE_ID = os.getenv("TOGGL_WORKSPACE_ID")
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
PG_HOST = os.getenv("PG_HOST", "localhost")
PG_PORT = int(os.getenv("PG_PORT", 5432))
PG_USER = os.getenv("PG_USER", "postgres")
PG_PASSWORD = os.getenv("PG_PASSWORD", None)
PG_DB = os.getenv("PG_DB", "time_tracking")

# Toggl 5.0 API base URL (canonical API endpoint)
TOGGL_BASE_URL = "https://api.track.toggl.com/api/v9"

# Initialize clients
def init_redis():
    try:
        client = redis.Redis(
            host=REDIS_HOST,
            port=REDIS_PORT,
            password=REDIS_PASSWORD,
            decode_responses=True,
            socket_connect_timeout=5
        )
        client.ping()
        logger.info("Redis client initialized")
        return client
    except redis.ConnectionError as e:
        logger.error(f"Redis connection failed: {str(e)}")
        raise SystemExit(1)

def init_pg():
    try:
        conn = psycopg2.connect(
            host=PG_HOST,
            port=PG_PORT,
            user=PG_USER,
            password=PG_PASSWORD,
            dbname=PG_DB
        )
        # Create table if not exists
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS toggl_time_entries (
                    id BIG PRIMARY KEY,
                    employee_id VARCHAR(255) NOT NULL,
                    billable_rate NUMERIC(10,2) NOT NULL,
                    duration_seconds INTEGER NOT NULL,
                    start_time TIMESTAMPTZ NOT NULL,
                    end_time TIMESTAMPTZ NOT NULL,
                    project_id BIG,
                    is_confirmed BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMPTZ DEFAULT NOW()
                );
                CREATE INDEX IF NOT EXISTS idx_toggl_employee_id ON toggl_time_entries(employee_id);
                CREATE INDEX IF NOT EXISTS idx_toggl_start_time ON toggl_time_entries(start_time);
            """)
        conn.commit()
        logger.info("PostgreSQL client initialized, table verified")
        return conn
    except psycopg2.Error as e:
        logger.error(f"PostgreSQL connection failed: {str(e)}")
        raise SystemExit(1)

def get_toggl_headers() -> Dict:
    """Return Toggl 5.0 API headers with authentication."""
    return {
        "Authorization": f"Basic {requests.auth._basic_auth_str(TOGGL_API_KEY, 'api_token')}",
        "Content-Type": "application/json",
        "User-Agent": "toggl-sync-100-devs/1.0"
    }

def fetch_all_employees(redis_client) -> List[str]:
    """Fetch all active employee IDs from Redis cache (populated by BambooHR webhook)."""
    keys = redis_client.keys("bamboo:employee:*")
    employee_ids = []
    for key in keys:
        try:
            data = json.loads(redis_client.get(key))
            if data.get("is_active", False):
                # Extract employee ID from cache key: bamboo:employee:{id}
                emp_id = key.split(":")[-1]
                employee_ids.append(emp_id)
        except (json.JSONDecodeError, redis.RedisError) as e:
            logger.warning(f"Failed to parse cache key {key}: {str(e)}")
            continue
    logger.info(f"Fetched {len(employee_ids)} active employees from cache")
    return employee_ids

def fetch_toggl_time_entries(employee_ids: List[str], start_date: datetime, end_date: datetime) -> List[Dict]:
    """Fetch time entries for all employees using Toggl 5.0 batch API.
    Batch endpoint supports 100 employee IDs per request, rate limit 500 req/min.
    """
    entries = []
    # Split employee IDs into batches of 100 (Toggl 5.0 batch limit)
    batches = [employee_ids[i:i+100] for i in range(0, len(employee_ids), 100)]
    logger.info(f"Fetching time entries for {len(employee_ids)} employees in {len(batches)} batches")

    for batch_idx, batch in enumerate(batches):
        url = f"{TOGGL_BASE_URL}/workspaces/{TOGGL_WORKSPACE_ID}/time_entries"
        params = {
            "start_date": start_date.isoformat(),
            "end_date": end_date.isoformat(),
            "employee_ids": ",".join(batch),
            "per_page": 1000  # Max per page for Toggl 5.0
        }
        try:
            response = requests.get(url, headers=get_toggl_headers(), params=params, timeout=10)
            # Handle rate limits: Toggl 5.0 returns 429 with Retry-After header
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 60))
                logger.warning(f"Rate limited, retrying after {retry_after} seconds")
                time.sleep(retry_after)
                response = requests.get(url, headers=get_toggl_headers(), params=params, timeout=10)
            response.raise_for_status()
            batch_entries = response.json()
            entries.extend(batch_entries)
            logger.info(f"Batch {batch_idx+1}/{len(batches)}: fetched {len(batch_entries)} entries")
            # Respect rate limit: 500 req/min → 1 req per 0.12s, but we're batching so no need
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to fetch batch {batch_idx+1}: {str(e)}")
            # Retry once on transient errors
            time.sleep(5)
            try:
                response = requests.get(url, headers=get_toggl_headers(), params=params, timeout=10)
                response.raise_for_status()
                entries.extend(response.json())
            except Exception as e2:
                logger.error(f"Retry failed for batch {batch_idx+1}: {str(e2)}")
                continue
    logger.info(f"Total fetched time entries: {len(entries)}")
    return entries

def enrich_entries_with_billable_rates(entries: List[Dict], redis_client) -> List[Dict]:
    """Enrich Toggl entries with billable rate from BambooHR Redis cache."""
    enriched = []
    for entry in entries:
        employee_id = str(entry.get("employee_id"))
        cache_key = f"bamboo:employee:{employee_id}"
        try:
            cache_data = redis_client.get(cache_key)
            if not cache_data:
                logger.warning(f"No BambooHR data for employee {employee_id}, using default rate 0")
                billable_rate = 0.0
            else:
                billable_data = json.loads(cache_data)
                billable_rate = float(billable_data.get("billable_rate", 0.0))
            entry["billable_rate"] = billable_rate
            enriched.append(entry)
        except (json.JSONDecodeError, redis.RedisError) as e:
            logger.error(f"Failed to enrich entry {entry.get('id')}: {str(e)}")
            continue
    return enriched

def write_entries_to_pg(entries: List[Dict], pg_conn):
    """Write enriched time entries to PostgreSQL warehouse."""
    with pg_conn.cursor() as cur:
        for entry in entries:
            try:
                cur.execute("""
                    INSERT INTO toggl_time_entries (id, employee_id, billable_rate, duration_seconds, start_time, end_time, project_id)
                    VALUES (%s, %s, %s, %s, %s, %s, %s)
                    ON CONFLICT (id) DO UPDATE SET
                        billable_rate = EXCLUDED.billable_rate,
                        duration_seconds = EXCLUDED.duration_seconds,
                        start_time = EXCLUDED.start_time,
                        end_time = EXCLUDED.end_time,
                        project_id = EXCLUDED.project_id;
                """, (
                    entry["id"],
                    str(entry["employee_id"]),
                    entry["billable_rate"],
                    entry["duration"],
                    datetime.fromisoformat(entry["start"]).replace(tzinfo=timezone.utc),
                    datetime.fromisoformat(entry["stop"]).replace(tzinfo=timezone.utc),
                    entry.get("project_id")
                ))
            except psycopg2.Error as e:
                logger.error(f"Failed to write entry {entry.get('id')}: {str(e)}")
                continue
        pg_conn.commit()
    logger.info(f"Wrote {len(entries)} entries to PostgreSQL")

if __name__ == "__main__":
    # Sync last 7 days of time entries
    end_date = datetime.now(timezone.utc)
    start_date = end_date - timedelta(days=7)
    logger.info(f"Starting Toggl sync from {start_date} to {end_date}")

    redis_client = init_redis()
    pg_conn = init_pg()

    try:
        employees = fetch_all_employees(redis_client)
        if not employees:
            logger.error("No active employees found in cache, exiting")
            raise SystemExit(1)
        raw_entries = fetch_toggl_time_entries(employees, start_date, end_date)
        enriched_entries = enrich_entries_with_billable_rates(raw_entries, redis_client)
        write_entries_to_pg(enriched_entries, pg_conn)
        logger.info("Sync completed successfully")
    except Exception as e:
        logger.error(f"Sync failed: {str(e)}")
        traceback.print_exc()
        raise
    finally:
        pg_conn.close()
        redis_client.close()

import flask
import requests
import redis
import json
import os
import logging
from datetime import datetime, timedelta, timezone
import hmac
import hashlib

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

app = flask.Flask(__name__)

# Load config
SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
SLACK_SIGNING_SECRET = os.getenv("SLACK_SIGNING_SECRET")
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
TOGGL_API_KEY = os.getenv("TOGGL_API_KEY")
TOGGL_WORKSPACE_ID = os.getenv("TOGGL_WORKSPACE_ID")
TOGGL_BASE_URL = "https://api.track.toggl.com/api/v9"

# Initialize Redis
try:
    redis_client = redis.Redis(
        host=REDIS_HOST,
        port=REDIS_PORT,
        password=REDIS_PASSWORD,
        decode_responses=True,
        socket_connect_timeout=5
    )
    redis_client.ping()
    logger.info("Redis client initialized")
except redis.ConnectionError as e:
    logger.error(f"Redis connection failed: {str(e)}")
    raise SystemExit(1)

def validate_slack_signature(request) -> bool:
    """Validate Slack 5.0 request signature to prevent forgery."""
    if not SLACK_SIGNING_SECRET:
        logger.warning("No SLACK_SIGNING_SECRET set, skipping validation")
        return True
    # Slack signature is HMAC-SHA256 of timestamp:body with signing secret
    timestamp = request.headers.get("X-Slack-Request-Timestamp")
    signature = request.headers.get("X-Slack-Signature")
    if not timestamp or not signature:
        logger.error("Missing Slack signature headers")
        return False
    # Check timestamp is within 5 minutes to prevent replay attacks
    if abs(int(datetime.now(timezone.utc).timestamp()) - int(timestamp)) > 300:
        logger.error("Slack request timestamp too old")
        return False
    # Compute expected signature
    body = request.get_data()
    sig_basestring = f"v0:{timestamp}:{body.decode('utf-8')}"
    expected_signature = "v0=" + hmac.new(
        SLACK_SIGNING_SECRET.encode("utf-8"),
        sig_basestring.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected_signature, signature)

def get_toggl_headers() -> Dict:
    return {
        "Authorization": f"Basic {requests.auth._basic_auth_str(TOGGL_API_KEY, 'api_token')}",
        "Content-Type": "application/json"
    }

def send_slack_approval_messages():
    """Send Friday 4pm Slack block kit messages to all active devs to confirm time entries."""
    # Get all active employees from Redis
    keys = redis_client.keys("bamboo:employee:*")
    employees = []
    for key in keys:
        try:
            data = json.loads(redis_client.get(key))
            if data.get("is_active", False):
                emp_id = key.split(":")[-1]
                # Get Slack user ID from BambooHR cache (assumes we store slack_id in BambooHR custom field)
                slack_id = data.get("slack_id")
                if slack_id:
                    employees.append({"emp_id": emp_id, "slack_id": slack_id})
        except Exception as e:
            logger.warning(f"Failed to parse {key}: {str(e)}")
            continue
    logger.info(f"Sending approval messages to {len(employees)} employees")

    for emp in employees:
        # Fetch unconfirmed time entries for the employee from Toggl
        url = f"{TOGGL_BASE_URL}/workspaces/{TOGGL_WORKSPACE_ID}/time_entries"
        params = {
            "employee_ids": emp["emp_id"],
            "start_date": (datetime.now(timezone.utc) - timedelta(days=7)).isoformat(),
            "end_date": datetime.now(timezone.utc).isoformat(),
            "per_page": 100
        }
        try:
            response = requests.get(url, headers=get_toggl_headers(), params=params, timeout=10)
            response.raise_for_status()
            entries = [e for e in response.json() if not e.get("is_confirmed", False)]
            if not entries:
                logger.info(f"No unconfirmed entries for {emp['emp_id']}, skipping")
                continue
            # Build Slack block kit 2.0 message (Slack 5.0 supports up to 4MB messages)
            blocks = [
                {
                    "type": "header",
                    "text": {"type": "plain_text", "text": "⏱️ Confirm Your Time Entries", "emoji": True}
                },
                {
                    "type": "section",
                    "text": {"type": "mrkdwn", "text": f"You have {len(entries)} unconfirmed time entries from the last 7 days. Please review and confirm below."}
                },
                {"type": "divider"},
            ]
            # Add each entry as a section with confirm button
            for entry in entries[:5]:  # Limit to 5 entries per message to avoid truncation
                start_time = datetime.fromisoformat(entry["start"]).strftime("%b %d %I:%M%p")
                duration_hours = entry["duration"] / 3600
                blocks.append({
                    "type": "section",
                    "text": {"type": "mrkdwn", "text": f"*{start_time}* – {entry.get('description', 'No description')} – {duration_hours:.2f} hours"},
                    "accessory": {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "Confirm", "emoji": True},
                        "style": "primary",
                        "value": json.dumps({"entry_id": entry["id"], "emp_id": emp["emp_id"]}),
                        "action_id": "confirm_time_entry"
                    }
                })
            blocks.append({"type": "divider"})
            blocks.append({
                "type": "section",
                "text": {"type": "mrkdwn", "text": "Need to edit an entry? Click below to open Toggl."},
                "accessory": {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Open Toggl", "emoji": True},
                    "url": f"https://track.toggl.com/timer"
                }
            })
            # Send message via Slack 5.0 chat.postMessage endpoint
            slack_response = requests.post(
                "https://slack.com/api/chat.postMessage",
                headers={
                    "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
                    "Content-Type": "application/json"
                },
                json={
                    "channel": emp["slack_id"],
                    "blocks": blocks,
                    "text": "Confirm your time entries"  # Fallback for clients that don't support blocks
                },
                timeout=10
            )
            slack_response.raise_for_status()
            if not slack_response.json().get("ok"):
                logger.error(f"Slack error for {emp['slack_id']}: {slack_response.json().get('error')}")
            else:
                logger.info(f"Sent approval message to {emp['slack_id']}")
        except Exception as e:
            logger.error(f"Failed to send message to {emp['emp_id']}: {str(e)}")
            continue

@app.route("/slack-interaction", methods=["POST"])
def handle_slack_interaction():
    """Handle Slack 5.0 interactive button clicks to confirm time entries."""
    # Validate signature
    if not validate_slack_signature(flask.request):
        return flask.jsonify({"error": "Invalid signature"}), 401
    # Parse payload (Slack sends URL-encoded payload)
    payload = json.loads(flask.request.form.get("payload"))
    action = payload["actions"][0]
    action_id = action["action_id"]
    if action_id != "confirm_time_entry":
        logger.warning(f"Unknown action ID: {action_id}")
        return flask.jsonify({"status": "ignored"}), 200
    # Extract entry ID and employee ID from button value
    try:
        value = json.loads(action["value"])
        entry_id = value["entry_id"]
        emp_id = value["emp_id"]
    except (KeyError, json.JSONDecodeError) as e:
        logger.error(f"Invalid action value: {str(e)}")
        return flask.jsonify({"error": "Invalid payload"}), 400
    # Update Toggl entry to confirmed
    url = f"{TOGGL_BASE_URL}/workspaces/{TOGGL_WORKSPACE_ID}/time_entries/{entry_id}"
    try:
        response = requests.put(
            url,
            headers=get_toggl_headers(),
            json={"is_confirmed": True},
            timeout=10
        )
        response.raise_for_status()
        logger.info(f"Confirmed entry {entry_id} for employee {emp_id}")
        # Update Slack message to show confirmed
        return flask.jsonify({
            "response_type": "ephemeral",
            "text": f"✅ Time entry {entry_id} confirmed successfully!"
        })
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to confirm entry {entry_id}: {str(e)}")
        return flask.jsonify({"error": "Failed to confirm entry"}), 500

if __name__ == "__main__":
    # Send approval messages every Friday at 4pm via cron, or run manually
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "--send":
        send_slack_approval_messages()
    else:
        app.run(host="0.0.0.0", port=8081, debug=False)

toggl-slack-bamboo-100-devs/
├── Dockerfile
├── requirements.txt
├── config/
│   ├── toggl.yaml
│   ├── slack.yaml
│   └── bamboo.yaml
├── src/
│   ├── bamboo_webhook.py  # BambooHR 5.0 webhook listener
│   ├── toggl_sync.py      # Toggl 5.0 time entry sync
│   ├── slack_approval.py  # Slack 5.0 interactive approval flow
│   └── utils/
│       ├── redis_client.py
│       ├── pg_client.py
│       └── signature_validator.py
├── tests/
│   ├── test_bamboo_webhook.py
│   ├── test_toggl_sync.py
│   └── test_slack_approval.py
└── README.md

# Batch fetch time entries for 100 employees
url = f"{TOGGL_BASE_URL}/workspaces/{TOGGL_WORKSPACE_ID}/time_entries"
params = {
    "start_date": start_date.isoformat(),
    "end_date": end_date.isoformat(),
    "employee_ids": ",".join(batch),  # Batch of 100 IDs
    "per_page": 1000
}
response = requests.get(url, headers=get_toggl_headers(), params=params)

def validate_slack_signature(request):
    timestamp = request.headers.get("X-Slack-Request-Timestamp")
    signature = request.headers.get("X-Slack-Signature")
    body = request.get_data()
    sig_basestring = f"v0:{timestamp}:{body.decode('utf-8')}"
    expected = "v0=" + hmac.new(SLACK_SIGNING_SECRET.encode(), sig_basestring.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

# Cache BambooHR employee data with 1 hour TTL
cache_key = f"bamboo:employee:{employee_id}"
cache_data = json.dumps({"billable_rate": 150.0, "role": "Senior Engineer", "is_active": True})
redis_client.setex(cache_key, 3600, cache_data)
# Read from cache
cached = redis_client.get(cache_key)
if cached:
    data = json.loads(cached)