Palantir CI Integration

Overview

Set up GitHub Actions CI pipelines for Foundry integrations. Covers running transform unit tests with PySpark, SDK integration tests with mocked APIs, and linting Foundry-specific patterns.

Prerequisites

• GitHub repository with Foundry integration code
• foundry-platform-sdk in requirements
• pytest test suite

Instructions

Step 1: GitHub Actions Workflow

# .github/workflows/foundry-ci.yml
name: Foundry CI
on:
  push:
    branches: [main]
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
          cache: pip

      - name: Set up Java (for PySpark)
        uses: actions/setup-java@v4
        with:
          distribution: temurin
          java-version: "11"

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run unit tests
        run: pytest tests/ -v --tb=short --junitxml=test-results.xml

      - name: Upload test results
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: test-results
          path: test-results.xml

  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install ruff
      - run: ruff check src/ tests/

  integration:
    runs-on: ubuntu-latest
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    needs: [test, lint]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -r requirements.txt
      - name: Run integration smoke test
        env:
          FOUNDRY_HOSTNAME: ${{ secrets.FOUNDRY_HOSTNAME }}
          FOUNDRY_CLIENT_ID: ${{ secrets.FOUNDRY_CLIENT_ID }}
          FOUNDRY_CLIENT_SECRET: ${{ secrets.FOUNDRY_CLIENT_SECRET }}
        run: python scripts/smoke_test.py

Step 2: Secret Configuration

# Add secrets to GitHub repository
gh secret set FOUNDRY_HOSTNAME --body "mycompany.palantirfoundry.com"
gh secret set FOUNDRY_CLIENT_ID --body "your-client-id"
gh secret set FOUNDRY_CLIENT_SECRET --body "your-client-secret"

Step 3: Custom Linting Rules for Foundry

# scripts/lint_foundry.py — catch common Foundry mistakes
import ast, sys

class FoundryLinter(ast.NodeVisitor):
    def visit_Str(self, node):
        # Flag hardcoded Foundry hostnames
        if "palantirfoundry.com" in node.s:
            print(f"  Line {node.lineno}: Hardcoded Foundry hostnam…

Palantir Common Errors

Overview

Quick reference for the top 10 most common Foundry API and transform errors with copy-paste solutions.

Prerequisites

• foundry-platform-sdk installed
• API credentials configured
• Access to Foundry build logs or application logs

Instructions

Error 1: 401 Unauthorized — Invalid or Expired Token

foundry.ApiError: 401 Unauthorized — The provided token is invalid or expired.

Fix:

# Regenerate token in Developer Console
# Settings > Tokens > Generate new personal access token
# Or re-authenticate with OAuth2:
auth = foundry.ConfidentialClientAuth(
    client_id=os.environ["FOUNDRY_CLIENT_ID"],
    client_secret=os.environ["FOUNDRY_CLIENT_SECRET"],
    hostname=os.environ["FOUNDRY_HOSTNAME"],
    scopes=["api:read-data"],
)
auth.sign_in_as_service_user()  # Gets a fresh token

Error 2: 403 Forbidden — Insufficient Scopes

foundry.ApiError: 403 Forbidden — Missing required scope: api:ontology-read

Fix: Add missing scopes in Developer Console > Your App > Scopes. Common scopes:

• api:read-data — read datasets
• api:write-data — write datasets
• api:ontology-read — read Ontology objects
• api:ontology-write — apply actions

Error 3: ObjectTypeNotFound

foundry.ApiError: 404 ObjectTypeNotFound — Object type 'employee' not found

Fix: Object type names are camelCase API names, not display names. Check Ontology Manager:

# List all object types to find the correct api_name
for ot in client.ontologies.ObjectType.list(ontology="my-company"):
    print(f"  {ot.api_name} (display: {ot.display_name})")

Error 4: DatasetNotFound

foundry.ApiError: 404 DatasetNotFound — Dataset not found or you do not have access

Fix: Verify the dataset RID (right-click dataset in Foundry UI > Copy RID). Ensure your service user has Viewer/Editor role on the project.

Error 5: Transform Build AnalysisException

pyspark.sql.utils.AnalysisException: cannot resolve 'fullname' given columns [fullName, department]

Fix: Spark column names are case-sensitive. Print columns to debug:

@transform_df(Output("/out"), data=Input("/in"))
def my_transform(data):
    print(data.columns)  # Check actual column names
    return data.select("fullName")  # Use exact casing

Error 6: OutOfMemoryError in Transform Builds

…

Palantir Core Workflow A — Data Pipelines with Transforms

Overview

Build Foundry data pipelines using the transforms-python library. Covers the @transform and @transform_df decorators, input/output dataset wiring, incremental transforms, and @configure for Spark tuning. This is the primary workflow for all data processing in Foundry.

Prerequisites

• Completed palantir-install-auth setup
• A Foundry Code Repository (Python Transforms type)
• Understanding of PySpark DataFrames (Foundry runs Spark under the hood)

Instructions

Step 1: Project Structure

my-transforms-repo/
├── src/
│   └── myproject/
│       ├── __init__.py
│       ├── pipeline.py          # Main transforms
│       ├── utils.py             # Shared logic
│       └── datasets.py          # Dataset path constants
├── build.gradle                 # Foundry build config
├── conda_recipe/meta.yaml       # Dependency declarations
└── settings.gradle

Step 2: Basic Transform with @transform_df

# src/myproject/pipeline.py
from transforms.api import transform_df, Input, Output

@transform_df(
    Output("/Company/datasets/cleaned_orders"),
    orders=Input("/Company/datasets/raw_orders"),
)
def clean_orders(orders):
    """Clean raw orders: drop nulls, normalize dates, filter test data."""
    from pyspark.sql import functions as F

    return (
        orders
        .filter(F.col("order_id").isNotNull())
        .filter(~F.col("email").like("%@test.com"))
        .withColumn("order_date", F.to_date("order_date_str", "yyyy-MM-dd"))
        .withColumn("total_cents", (F.col("total") * 100).cast("long"))
        .drop("order_date_str", "total")
    )

Step 3: Multi-Input Join Transform

@transform_df(
    Output("/Company/datasets/order_enriched"),
    orders=Input("/Company/datasets/cleaned_orders"),
    customers=Input("/Company/datasets/customers"),
)
def enrich_orders(orders, customers):
    """Join orders with customer data for analytics."""
    from pyspark.sql import functions as F

    return (
        orders
        .join(customers, orders.customer_id == customers.id, "left")
        .select(
            orders.order_id,
            orders.order_date,
            orders.total_cents,
            customers.name.alias("customer_name"),
            customers.segment,
            customers.region,
        )
        .withColumn("processed_at", F.current_timestamp())
    )

Step 4: Low-Level @transform for File I/O

from…

Palantir Core Workflow B — Ontology Objects & Actions

Overview

Query, filter, and mutate Ontology objects using the Foundry Platform SDK and OSDK. Covers listing objects with property filters, following links between object types, applying actions, and aggregating object data. This is the primary workflow for Ontology-driven applications.

Prerequisites

• Completed palantir-install-auth setup
• An Ontology with configured object types, link types, and actions
• Familiarity with palantir-core-workflow-a (data pipelines feed the Ontology)

Instructions

Step 1: List and Filter Objects (REST API)

import os, foundry

client = foundry.FoundryClient(
    auth=foundry.UserTokenAuth(
        hostname=os.environ["FOUNDRY_HOSTNAME"],
        token=os.environ["FOUNDRY_TOKEN"],
    ),
    hostname=os.environ["FOUNDRY_HOSTNAME"],
)

ONTOLOGY = "my-company"

# List employees in Engineering, sorted by hire date
result = client.ontologies.OntologyObject.list(
    ontology=ONTOLOGY,
    object_type="Employee",
    page_size=20,
    order_by="hireDate:asc",
    properties={"department": "Engineering"},
)

for obj in result.data:
    p = obj.properties
    print(f"{p['fullName']} | {p['department']} | hired {p['hireDate']}")

Step 2: Search Objects with Filters

# Search with complex filters using the search endpoint
search_result = client.ontologies.OntologyObject.search(
    ontology=ONTOLOGY,
    object_type="Employee",
    where={
        "type": "and",
        "value": [
            {"type": "eq", "field": "department", "value": "Engineering"},
            {"type": "gte", "field": "salary", "value": 100000},
        ],
    },
    page_size=50,
)
print(f"Found {len(search_result.data)} matching employees")

Step 3: Follow Links Between Objects

# Get all projects linked to an employee
employee_rid = "ri.ontology.main.object.employee-001"

linked_projects = client.ontologies.OntologyObject.list_linked_objects(
    ontology=ONTOLOGY,
    object_type="Employee",
    primary_key="EMP-001",
    link_type="assignedProjects",
)

for project in linked_projects.data:
    print(f"  Project: {project.properties['name']} — {project.properties['status']}")

Step 4: Apply Actions to Modify Objects

# Promote an employee — triggers validation rules defined in Ontology
result = client.ontologies.Action.apply(
    ontology=ONTOLOGY,
    a…

Palantir Cost Tuning

Overview

Optimize Foundry compute and API costs through incremental transforms, right-sized Spark profiles, efficient pagination, and usage monitoring.

Prerequisites

• Active Foundry enrollment with build history
• Access to Foundry resource usage metrics
• Understanding of transform build patterns

Instructions

Step 1: Cost Drivers in Foundry

Step 2: Convert Full Rebuilds to Incremental

from transforms.api import transform_df, Input, Output, incremental

# BEFORE: Full rebuild every run (expensive for large datasets)
@transform_df(Output("/out"), data=Input("/in"))
def expensive(data):
    return data.filter(data.status == "active")

# AFTER: Only processes new/changed rows
@incremental()
@transform_df(Output("/out"), data=Input("/in"))
def cheap(data):
    return data.filter(data.status == "active")

Step 3: Right-Size Spark Profiles

from transforms.api import configure

# DON'T: Default profile for everything
# DO: Match profile to actual data size

# Small data (< 1GB) — use lightweight transforms (no Spark)
from transforms.api import transform_polars
@transform_polars(Output("/out"), data=Input("/small_table"))
def small_job(data):
    return data.filter(data["status"] == "active")

# Medium data (1-50GB) — default profile is fine
@transform_df(Output("/out"), data=Input("/medium_table"))
def medium_job(data):
    return data.select("id", "name")

# Large data (50GB+) — explicit large profile
@configure(profile=["DRIVER_MEMORY_LARGE"])
@transform_df(Output("/out"), data=Input("/big_table"))
def large_job(data):
    return data.groupBy("region").count()

Step 4: Replace Polling with Webhooks

# EXPENSIVE: Polling every 30 seconds
import time
while True:
    result = client.ontologies.OntologyObject.list(
        ontology="co", object_type="Order", page_size=100,
    )
    process_new_orders(result.data)
    time.sleep(30)  # 2,880 API calls/day!

# CHEAP: Webhook-driven (0 pol…

Palantir Data Handling

Overview

Handle sensitive data in Foundry using markings (data classifications), column-level security, PII redaction in transforms, and GDPR/CCPA deletion workflows.

Prerequisites

• Foundry enrollment with Markings enabled
• Understanding of your organization's data classification policy
• Familiarity with transforms (palantir-core-workflow-a)

Instructions

Step 1: Data Classification with Markings

Foundry Markings control who can access data at the dataset, column, or row level.

Step 2: PII Redaction in Transforms

from transforms.api import transform_df, Input, Output
from pyspark.sql import functions as F

@transform_df(
    Output("/Company/datasets/customers_safe"),
    customers=Input("/Company/datasets/raw_customers"),
)
def redact_pii(customers):
    """Create an analytics-safe view with PII removed."""
    return (
        customers
        .withColumn("email", F.sha2(F.col("email"), 256))           # Hash email
        .withColumn("phone", F.lit("***-***-****"))                   # Mask phone
        .withColumn("ssn", F.lit(None).cast("string"))                # Remove SSN
        .withColumn("name", F.concat(
            F.substring("first_name", 1, 1), F.lit("***")            # First initial only
        ))
        .drop("first_name", "last_name", "address", "date_of_birth")
    )

Step 3: GDPR Right to Erasure

def delete_user_data(client, user_id: str):
    """GDPR Article 17: delete all data for a specific user."""
    datasets_with_pii = [
        "/Company/datasets/raw_customers",
        "/Company/datasets/raw_orders",
        "/Company/datasets/customer_communications",
    ]
    for dataset_path in datasets_with_pii:
        # Trigger a transform that filters out the user
        client.ontologies.Action.apply(
            ontology="my-company",
            action_type="gdprDeleteUser",
            parameters={"userId": user_id, "datasetPath": dataset_path},
        )
    # Log the deletion …

Palantir Debug Bundle

Overview

Collect all diagnostic information needed for Foundry support tickets: SDK version, auth status, API connectivity, build logs, and environment details. Secrets are automatically redacted.

Prerequisites

• foundry-platform-sdk installed
• Access to application logs and Foundry build logs
• Permission to collect environment info

Instructions

Step 1: Create Debug Bundle Script

#!/bin/bash
set -euo pipefail
BUNDLE_DIR="foundry-debug-$(date +%Y%m%d-%H%M%S)"
mkdir -p "$BUNDLE_DIR"

echo "=== Foundry Debug Bundle ===" > "$BUNDLE_DIR/summary.txt"
echo "Generated: $(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$BUNDLE_DIR/summary.txt"

# Python environment
echo -e "\n--- Python Environment ---" >> "$BUNDLE_DIR/summary.txt"
python --version >> "$BUNDLE_DIR/summary.txt" 2>&1
pip show foundry-platform-sdk 2>/dev/null | grep -E "^(Name|Version)" >> "$BUNDLE_DIR/summary.txt"
pip show palantir-sdk 2>/dev/null | grep -E "^(Name|Version)" >> "$BUNDLE_DIR/summary.txt"

# Environment variables (redacted)
echo -e "\n--- Environment (redacted) ---" >> "$BUNDLE_DIR/summary.txt"
echo "FOUNDRY_HOSTNAME: ${FOUNDRY_HOSTNAME:-NOT SET}" >> "$BUNDLE_DIR/summary.txt"
echo "FOUNDRY_TOKEN: ${FOUNDRY_TOKEN:+[SET, length=${#FOUNDRY_TOKEN}]}" >> "$BUNDLE_DIR/summary.txt"
echo "FOUNDRY_CLIENT_ID: ${FOUNDRY_CLIENT_ID:+[SET]}" >> "$BUNDLE_DIR/summary.txt"
echo "FOUNDRY_CLIENT_SECRET: ${FOUNDRY_CLIENT_SECRET:+[SET]}" >> "$BUNDLE_DIR/summary.txt"

Step 2: Test API Connectivity

# API connectivity test
echo -e "\n--- API Connectivity ---" >> "$BUNDLE_DIR/summary.txt"
if [ -n "${FOUNDRY_HOSTNAME:-}" ] && [ -n "${FOUNDRY_TOKEN:-}" ]; then
  HTTP_CODE=$(curl -s -o "$BUNDLE_DIR/api-response.json" -w "%{http_code}" \
    -H "Authorization: Bearer $FOUNDRY_TOKEN" \
    "https://$FOUNDRY_HOSTNAME/api/v2/ontologies" 2>/dev/null || echo "FAILED")
  echo "Ontology API: HTTP $HTTP_CODE" >> "$BUNDLE_DIR/summary.txt"
else
  echo "Skipped: FOUNDRY_HOSTNAME or FOUNDRY_TOKEN not set" >> "$BUNDLE_DIR/summary.txt"
fi

Step 3: Collect Build Logs and Error Context

# Collect recent Python errors
echo -e "\n--- Recent Errors ---" >> "$BUNDLE_DIR/summary.txt"
grep -rn "foundry\.\|ApiError\|Traceback" *.log 2>/dev/null | tail -30 >> "$BUNDLE_DIR/erro…

Palantir Deploy Integration

Overview

Deploy Foundry-integrated applications to cloud platforms (GCP Cloud Run, AWS Lambda, Docker) with proper secrets management and health checks.

Prerequisites

• Passing CI tests: palantir-ci-integration
• Production OAuth2 credentials from Developer Console
• Cloud platform CLI configured (gcloud, aws, etc.)

Instructions

Step 1: Dockerfile

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ ./src/
EXPOSE 8080
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8080"]

Step 2: Deploy to Google Cloud Run

set -euo pipefail
PROJECT_ID=$(gcloud config get-value project)
SERVICE_NAME="foundry-integration"
REGION="us-central1"

# Build and push container
gcloud builds submit --tag "gcr.io/$PROJECT_ID/$SERVICE_NAME"

# Deploy with secrets from Secret Manager
gcloud run deploy "$SERVICE_NAME" \
  --image "gcr.io/$PROJECT_ID/$SERVICE_NAME" \
  --region "$REGION" \
  --set-secrets "FOUNDRY_HOSTNAME=foundry-hostname:latest" \
  --set-secrets "FOUNDRY_CLIENT_ID=foundry-client-id:latest" \
  --set-secrets "FOUNDRY_CLIENT_SECRET=foundry-client-secret:latest" \
  --min-instances 1 \
  --max-instances 10 \
  --timeout 60 \
  --allow-unauthenticated

Step 3: Health Check Endpoint

# src/main.py
from fastapi import FastAPI
import foundry, os

app = FastAPI()

@app.get("/health")
async def health():
    try:
        client = get_foundry_client()
        list(client.ontologies.Ontology.list())
        return {"status": "healthy", "foundry": "connected"}
    except Exception as e:
        return {"status": "degraded", "foundry": str(e)}, 503

Step 4: Environment-Specific Configuration

# src/config.py
import os
from dataclasses import dataclass

@dataclass
class FoundryConfig:
    hostname: str
    client_id: str
    client_secret: str
    scopes: list[str]

    @classmethod
    def from_env(cls) -> "FoundryConfig":
        env = os.environ.get("ENVIRONMENT", "development")
        scopes_map = {
            "development": ["api:read-data"],
            "staging": ["api:read-data", "api:write-data"],
            "production": ["api:read-data", "api:write-data", "api:ontology-read"],
        }
        return cls(
            hostname=os.environ["FOUNDRY_HOSTNAME"],
            client_id=os.environ["…

Palantir Enterprise RBAC

Overview

Configure enterprise-grade access control in Foundry: project roles (Viewer/Editor/Owner), organization-level groups, service user accounts for integrations, and marking-based data classification.

Prerequisites

• Foundry enrollment with admin access
• Understanding of Foundry project structure
• Familiarity with palantir-security-basics

Instructions

Step 1: Project Role Hierarchy

Step 2: Create Service Users for Integrations

Developer Console > Applications > New Application:
1. Name: "order-sync-service" (descriptive of function)
2. Type: Server application (client credentials flow)
3. Scopes: api:read-data, api:ontology-read (minimum needed)
4. Project access: Add as Editor on specific projects only

Result: client_id + client_secret (store in secrets manager)

Step 3: Scope Matrix by Application

# Define per-application scopes
APP_SCOPES = {
    "dashboard-reader": ["api:read-data", "api:ontology-read"],
    "data-sync-service": ["api:read-data", "api:write-data"],
    "admin-tool": ["api:read-data", "api:write-data", "api:ontology-read", "api:ontology-write"],
}

def create_client_for_app(app_name: str) -> foundry.FoundryClient:
    scopes = APP_SCOPES[app_name]
    auth = foundry.ConfidentialClientAuth(
        client_id=os.environ[f"{app_name.upper().replace('-','_')}_CLIENT_ID"],
        client_secret=os.environ[f"{app_name.upper().replace('-','_')}_CLIENT_SECRET"],
        hostname=os.environ["FOUNDRY_HOSTNAME"],
        scopes=scopes,
    )
    auth.sign_in_as_service_user()
    return foundry.FoundryClient(auth=auth, hostname=os.environ["FOUNDRY_HOSTNAME"])

Step 4: Group-Based Access Control

Organization Groups (manage in Foundry Admin):
├── data-engineering        → Editor on pipeline projects
├── data-science            → Viewer on pipeline, Editor on ML projects
├── business-analysts       → Viewer on analytics projects
├── external-partners       → Viewer on shared datasets only
└── platform-admins         → Owner on all projects

Principle: Users inherit access from groups.
Never assign project roles to individual users.

Step…

Palantir Hello World

Overview

Build a minimal working example that connects to Palantir Foundry, queries Ontology objects via the REST API, reads a dataset, and applies an action. Uses real foundry-platform-sdk Python API calls.

Prerequisites

• Completed palantir-install-auth setup
• Valid bearer token or OAuth2 credentials
• At least one Ontology with object types configured in your Foundry enrollment

Instructions

Step 1: List Available Ontologies

import os
import foundry

client = foundry.FoundryClient(
    auth=foundry.UserTokenAuth(
        hostname=os.environ["FOUNDRY_HOSTNAME"],
        token=os.environ["FOUNDRY_TOKEN"],
    ),
    hostname=os.environ["FOUNDRY_HOSTNAME"],
)

# List all ontologies you have access to
for ont in client.ontologies.Ontology.list():
    print(f"Ontology: {ont.api_name}  RID: {ont.rid}")

Step 2: Query Ontology Objects

# List objects of type "Employee" from the default ontology
# The object type api_name comes from your Ontology configuration
ONTOLOGY = "your-ontology-api-name"
OBJECT_TYPE = "Employee"

objects = client.ontologies.OntologyObject.list(
    ontology=ONTOLOGY,
    object_type=OBJECT_TYPE,
    page_size=5,
)

for obj in objects.data:
    props = obj.properties
    print(f"  {props.get('fullName', 'N/A')} — {props.get('department', 'N/A')}")

Step 3: Get a Single Object by Primary Key

employee = client.ontologies.OntologyObject.get(
    ontology=ONTOLOGY,
    object_type=OBJECT_TYPE,
    primary_key="EMP-001",
)
print(f"Found: {employee.properties}")

Step 4: Read a Dataset

# Read rows from a Foundry dataset (tabular)
DATASET_RID = "ri.foundry.main.dataset.xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

# Get dataset metadata
dataset = client.datasets.Dataset.get(dataset_rid=DATASET_RID)
print(f"Dataset: {dataset.name}, Path: {dataset.path}")

# Read rows from the dataset (CSV format)
content = client.datasets.Dataset.read(
    dataset_rid=DATASET_RID,
    branch_id="master",
    format="arrow",  # or "csv"
)
print(f"Read {len(content)} bytes of data")

Step 5: Apply an Ontology Action

# Actions modify objects — e.g., updating an employee's department
result = client.ontologies.Action.apply(
    ontology=ONTOLOGY,
    action_type="updateDepartment",
    parameters={
        "employeeId": "EMP-001",
        "newDepartment": "Engineering",
    },
)
print(f"Action result: {result.validation}")

Palantir Incident Runbook

Overview

Rapid incident response for Foundry-related outages: API failures, transform build failures, authentication issues, and data pipeline stalls.

Prerequisites

• Access to application logs and Foundry build history
• Foundry service user credentials for health checks
• On-call escalation path defined

Instructions

Step 1: Triage (First 5 Minutes)

set -euo pipefail
echo "=== Foundry Incident Triage ==="
echo "Time: $(date -u)"

# 1. Check if Foundry itself is down
curl -s -o /dev/null -w "Foundry API: HTTP %{http_code}\n" \
  -H "Authorization: Bearer $FOUNDRY_TOKEN" \
  "https://$FOUNDRY_HOSTNAME/api/v2/ontologies" || echo "FOUNDRY UNREACHABLE"

# 2. Check our app health
curl -s http://localhost:8080/health | python -m json.tool

# 3. Check recent error logs
grep -c "ApiError\|status_code.*[45][0-9][0-9]" /var/log/app/app.log | tail -1

Step 2: Classify Severity

Step 3: Common Incident Playbooks

Playbook A: Authentication Failure (401/403)

# 1. Verify token is set
echo "Token set: ${FOUNDRY_TOKEN:+yes}"
echo "Token length: ${#FOUNDRY_TOKEN}"

# 2. Test with a fresh token
python -c "
import os, foundry
client = foundry.FoundryClient(
    auth=foundry.UserTokenAuth(
        hostname=os.environ['FOUNDRY_HOSTNAME'],
        token=os.environ['FOUNDRY_TOKEN'],
    ),
    hostname=os.environ['FOUNDRY_HOSTNAME'],
)
print('Auth OK:', list(client.ontologies.Ontology.list())[0].api_name)
"
# 3. If still failing: regenerate credentials in Developer Console

Playbook B: Rate Limiting (429)

# 1. Check rate limit headers from last response
# 2. Enable request throttling
# 3. Review batch operations for unnecessary API calls
# See palantir-rate-limits for detailed implementation

Playbook C: Transform Build Failure

1. Open Foundry > Pipeline Builder > failed build
2. Check the "Errors" tab for stack trace
3. Common causes:
   - OutOfMemoryError → add @configure(profile=["DRIVER_MEMORY_LARGE…

Palantir Install & Auth

Overview

Set up the Palantir Foundry Platform SDK (Python or TypeScript) and configure authentication using either bearer tokens for development or OAuth2 client credentials for production. Covers both the Platform SDK for direct API access and the OSDK for Ontology-based workflows.

Prerequisites

• Python 3.9+ or Node.js 18+
• A Palantir Foundry enrollment with API access enabled
• A third-party application registered in Developer Console (for OAuth2)
• Your Foundry hostname (e.g., mycompany.palantirfoundry.com)

Instructions

Step 1: Install the SDK

Python (Platform SDK):

set -euo pipefail
pip install foundry-platform-sdk
python -c "import foundry; print(f'foundry-platform-sdk {foundry.__version__} installed')"

Python (OSDK for Ontology access):

set -euo pipefail
pip install palantir-sdk
python -c "import palantir; print('palantir-sdk installed')"

TypeScript (OSDK):

set -euo pipefail
npm install @osdk/client @osdk/oauth
npx tsc --version

Step 2: Configure Environment Variables

# .env — never commit this file
FOUNDRY_HOSTNAME=mycompany.palantirfoundry.com

# Option A: Bearer token (development only)
FOUNDRY_TOKEN=eyJhbGciOiJS...

# Option B: OAuth2 client credentials (production)
FOUNDRY_CLIENT_ID=abc123
FOUNDRY_CLIENT_SECRET=secret456

Step 3: Initialize with Bearer Token (Development)

import os
import foundry

client = foundry.FoundryClient(
    auth=foundry.UserTokenAuth(
        hostname=os.environ["FOUNDRY_HOSTNAME"],
        token=os.environ["FOUNDRY_TOKEN"],
    ),
    hostname=os.environ["FOUNDRY_HOSTNAME"],
)

# Verify connection — list datasets
datasets = client.datasets.Dataset.list()
for ds in datasets:
    print(f"Dataset: {ds.rid}")

Step 4: Initialize with OAuth2 (Production)

import os
import foundry

auth = foundry.ConfidentialClientAuth(
    client_id=os.environ["FOUNDRY_CLIENT_ID"],
    client_secret=os.environ["FOUNDRY_CLIENT_SECRET"],
    hostname=os.environ["FOUNDRY_HOSTNAME"],
    scopes=["api:read-data", "api:write-data"],
)
auth.sign_in_as_service_user()

client = foundry.FoundryClient(
    auth=auth,
    hostname=os.environ["FOUNDRY_HOSTNAME"],
)

# Verify — fetch ontology metadata
ontologies = client.ontologies.Ontology.list()
for ont in ontologies:
    print(f"Ontology: {ont.api_name} (rid={ont.rid})")

Step 5: TypeScript OSDK Setup

import {…

Palantir Local Dev Loop

Overview

Set up local development for Palantir Foundry integrations. Covers running transforms locally against sample data, mocking the Foundry API for fast iteration, and testing with pytest before pushing to Foundry.

Prerequisites

• Completed palantir-install-auth setup
• Python 3.9+ with pip
• A Foundry Code Repository cloned locally (or a standalone project)

Instructions

Step 1: Project Structure

my-foundry-project/
├── src/myproject/
│   ├── __init__.py
│   ├── pipeline.py          # @transform functions
│   └── utils.py             # Shared logic
├── tests/
│   ├── conftest.py           # Fixtures with sample DataFrames
│   ├── test_pipeline.py      # Transform unit tests
│   └── sample_data/          # CSV/Parquet test fixtures
├── .env                      # FOUNDRY_HOSTNAME, FOUNDRY_TOKEN
├── requirements.txt          # foundry-platform-sdk, pytest, pyspark
└── pyproject.toml

Step 2: Install Local Dependencies

set -euo pipefail
pip install foundry-platform-sdk pyspark pytest pandas
python -c "import foundry; import pyspark; print('Dependencies ready')"

Step 3: Test Transforms Locally with PySpark

# tests/conftest.py
import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder.master("local[2]").appName("test").getOrCreate()

@pytest.fixture
def sample_orders(spark):
    data = [
        ("ORD-001", "alice@company.com", "2026-03-01", 99.99),
        ("ORD-002", "bob@test.com", "2026-03-02", 49.99),      # test email
        (None, "carol@company.com", "2026-03-03", 149.99),       # null ID
    ]
    return spark.createDataFrame(data, ["order_id", "email", "order_date_str", "total"])

# tests/test_pipeline.py
from myproject.pipeline import clean_orders

def test_clean_orders_removes_nulls_and_test_emails(sample_orders):
    result = clean_orders(sample_orders)
    assert result.count() == 1  # Only alice remains
    assert result.columns == ["order_id", "email", "order_date", "total_cents"]
    row = result.first()
    assert row.total_cents == 9999  # 99.99 * 100

Step 4: Mock Foundry API for Integration Tests

# tests/test_api.py
import pytest
from unittest.mock import MagicMock, patch

def test_list_ontology_objects():
    mock_client = MagicMock()
    mock_client.ontologies.OntologyObject.list.return_value.data = [
        MagicMock(properties={"fullName": "Alice", "department": "Enginee…

Palantir Migration Deep Dive

Overview

Comprehensive guide for migrating data into Foundry, migrating from legacy systems to Foundry-backed architectures, and upgrading between Foundry API versions using the strangler fig pattern.

Prerequisites

• Source system access and schema documentation
• Foundry enrollment with write access
• Understanding of Foundry data pipeline architecture (palantir-reference-architecture)

Instructions

Step 1: Migration Assessment

## Migration Checklist
- [ ] Source system inventory (tables, volumes, refresh rates)
- [ ] Data classification (PII, confidential, public)
- [ ] Schema mapping: source columns → Foundry dataset columns
- [ ] Volume estimate: rows, GB, growth rate
- [ ] Dependencies: downstream consumers of source data
- [ ] Timeline: parallel run period, cutover date

Step 2: Data Migration — Bulk Import

import foundry, pandas as pd

client = get_foundry_client()

# Read source data (example: PostgreSQL)
df = pd.read_sql("SELECT * FROM orders WHERE year >= 2024", source_conn)

# Upload to Foundry dataset
client.datasets.Dataset.upload(
    dataset_rid="ri.foundry.main.dataset.xxxxx",
    branch_id="master",
    file_path="orders.parquet",
    data=df.to_parquet(),
    content_type="application/x-parquet",
)
print(f"Uploaded {len(df)} rows to Foundry")

Step 3: Incremental Sync (Ongoing)

from datetime import datetime, timedelta

def incremental_sync(client, source_conn, dataset_rid, last_sync):
    """Sync only new/changed rows since last sync."""
    query = f"""
        SELECT * FROM orders 
        WHERE updated_at > '{last_sync.isoformat()}'
        ORDER BY updated_at
    """
    df = pd.read_sql(query, source_conn)
    if df.empty:
        print("No new rows to sync")
        return last_sync

    client.datasets.Dataset.upload(
        dataset_rid=dataset_rid,
        branch_id="master",
        file_path=f"sync_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.parquet",
        data=df.to_parquet(),
    )
    print(f"Synced {len(df)} rows")
    return df["updated_at"].max()

Step 4: Strangler Fig Pattern for API Migration

class DualWriteClient:
    """Write to both legacy and Foundry during migration period."""
    def __init__(self, legacy_client, foundry_client):
        self.legacy = legacy_client
        self.foundry = foundry_client
        self.foundry_enabled = os.environ.get("FOUNDRY_WRITES_ENABLED", "false") == "true"

    def create_order(self, order_da…

Palantir Multi-Environment Setup

Overview

Configure Foundry integrations across dev/staging/prod environments with separate credentials, enrollment hostnames, and scope policies per environment.

Prerequisites

• Foundry enrollments for each environment (or separate projects within one enrollment)
• Secrets manager (AWS SM, GCP SM, or Vault)
• Familiarity with palantir-security-basics

Instructions

Step 1: Environment Configuration

# src/config.py
import os
from dataclasses import dataclass

@dataclass
class FoundryEnvConfig:
    hostname: str
    client_id: str
    client_secret: str
    scopes: list[str]
    ontology: str

ENVIRONMENTS = {
    "development": FoundryEnvConfig(
        hostname=os.environ.get("DEV_FOUNDRY_HOSTNAME", "dev.palantirfoundry.com"),
        client_id=os.environ.get("DEV_FOUNDRY_CLIENT_ID", ""),
        client_secret=os.environ.get("DEV_FOUNDRY_CLIENT_SECRET", ""),
        scopes=["api:read-data"],  # Read-only in dev
        ontology="dev-ontology",
    ),
    "staging": FoundryEnvConfig(
        hostname=os.environ.get("STG_FOUNDRY_HOSTNAME", "staging.palantirfoundry.com"),
        client_id=os.environ.get("STG_FOUNDRY_CLIENT_ID", ""),
        client_secret=os.environ.get("STG_FOUNDRY_CLIENT_SECRET", ""),
        scopes=["api:read-data", "api:write-data"],
        ontology="staging-ontology",
    ),
    "production": FoundryEnvConfig(
        hostname=os.environ.get("PROD_FOUNDRY_HOSTNAME", "prod.palantirfoundry.com"),
        client_id=os.environ.get("PROD_FOUNDRY_CLIENT_ID", ""),
        client_secret=os.environ.get("PROD_FOUNDRY_CLIENT_SECRET", ""),
        scopes=["api:read-data", "api:write-data", "api:ontology-read"],
        ontology="production-ontology",
    ),
}

def get_config() -> FoundryEnvConfig:
    env = os.environ.get("ENVIRONMENT", "development")
    return ENVIRONMENTS[env]

Step 2: Environment-Aware Client Factory

import foundry

def create_client(config: FoundryEnvConfig) -> foundry.FoundryClient:
    auth = foundry.ConfidentialClientAuth(
        client_id=config.client_id,
        client_secret=config.client_secret,
        hostname=config.hostname,
        scopes=config.scopes,
    )
    auth.sign_in_as_service_user()
    return foundry.FoundryClient(auth=auth, hostname=config.hostname)

# Usage
config = get_config()
client = create_client(config)

Step 3: Environment Variables per Platform

# Docker Compose
# docker-compose.yml
services…

Palantir Observability

Overview

Set up comprehensive observability for Foundry integrations: structured logging with request IDs, Prometheus metrics for API latency/errors, health check endpoints, and alert rules.

Prerequisites

• Working Foundry integration
• Prometheus + Grafana (or equivalent monitoring stack)
• Familiarity with palantir-prod-checklist

Instructions

Step 1: Structured Logging

import logging, json, time, uuid

class FoundryLogger:
    def __init__(self):
        self.logger = logging.getLogger("foundry")
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter("%(message)s"))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_api_call(self, method: str, endpoint: str, status: int, duration_ms: float):
        self.logger.info(json.dumps({
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "request_id": str(uuid.uuid4())[:8],
            "service": "foundry",
            "method": method,
            "endpoint": endpoint,
            "status": status,
            "duration_ms": round(duration_ms, 2),
            "level": "error" if status >= 400 else "info",
        }))

Step 2: Prometheus Metrics

from prometheus_client import Counter, Histogram, Gauge

foundry_requests = Counter(
    "foundry_api_requests_total",
    "Total Foundry API requests",
    ["method", "endpoint", "status"],
)
foundry_latency = Histogram(
    "foundry_api_latency_seconds",
    "Foundry API request latency",
    ["endpoint"],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
)
foundry_health = Gauge(
    "foundry_api_healthy",
    "1 if Foundry API is reachable, 0 otherwise",
)

def instrumented_call(client, method, *args, **kwargs):
    endpoint = method.__qualname__
    start = time.monotonic()
    try:
        result = method(*args, **kwargs)
        status = 200
        return result
    except foundry.ApiError as e:
        status = e.status_code
        raise
    finally:
        duration = time.monotonic() - start
        foundry_requests.labels(method="API", endpoint=endpoint, status=str(status)).inc()
        foundry_latency.labels(endpoint=endpoint).observe(duration)

Step 3: Health Check with Metrics

import time

async def foundry_health_check():
    start = time.monotonic()
    try:
        list(client.ontologies.Ontology.list())
        latency = (time.monotonic() - start) * 1000
        foundry_health.set(1)
        return {"status…

Palantir Performance Tuning

Overview

Optimize Foundry API performance: efficient pagination, client-side caching, batch object retrieval, and Spark transform tuning with @configure profiles.

Prerequisites

• Completed palantir-install-auth setup
• Working Foundry integration to optimize
• Access to Foundry build metrics (for transform tuning)

Instructions

Step 1: Efficient Pagination

from functools import lru_cache

def fetch_all_objects(client, ontology: str, object_type: str, page_size: int = 500):
    """Fetch all objects with maximum page size to minimize API calls."""
    all_objects = []
    page_token = None
    while True:
        result = client.ontologies.OntologyObject.list(
            ontology=ontology,
            object_type=object_type,
            page_size=min(page_size, 500),  # Foundry max is 500
            page_token=page_token,
        )
        all_objects.extend(result.data)
        page_token = result.next_page_token
        if not page_token:
            break
    return all_objects

Step 2: Client-Side Caching

from cachetools import TTLCache
import hashlib, json

_cache = TTLCache(maxsize=1000, ttl=300)  # 5-minute TTL

def cached_get_object(client, ontology, object_type, primary_key):
    """Cache Ontology object reads to reduce API calls."""
    cache_key = f"{ontology}:{object_type}:{primary_key}"
    if cache_key in _cache:
        return _cache[cache_key]
    obj = client.ontologies.OntologyObject.get(
        ontology=ontology, object_type=object_type, primary_key=primary_key,
    )
    _cache[cache_key] = obj
    return obj

def invalidate_cache(ontology, object_type, primary_key):
    cache_key = f"{ontology}:{object_type}:{primary_key}"
    _cache.pop(cache_key, None)

Step 3: Batch Object Retrieval

def batch_get_objects(client, ontology, object_type, primary_keys, batch_size=50):
    """Retrieve multiple objects using search filter instead of individual GETs."""
    results = {}
    for i in range(0, len(primary_keys), batch_size):
        batch = primary_keys[i:i+batch_size]
        search_result = client.ontologies.OntologyObject.search(
            ontology=ontology,
            object_type=object_type,
            where={
                "type": "in",
                "field": "primaryKey",
                "value": batch,
            },
            page_size=batch_size,
        )
        for obj in search_result.data:
            pk = obj.properties.get("primaryKey", obj.rid)
            results[pk] = obj
    return results

Step 4: Transform Build Performance

Palantir Production Checklist

Overview

Complete go-live checklist for deploying Foundry-integrated applications to production. Covers credential management, health checks, monitoring, and rollback procedures.

Prerequisites

• Staging environment tested and verified
• Production OAuth2 credentials from Developer Console
• Deployment pipeline configured
• Monitoring infrastructure ready

Instructions

Pre-Deployment: Credentials & Config

• [ ] OAuth2 client credentials in secrets manager (not personal tokens)
• [ ] Scopes are minimal: only what the app actually needs
• [ ] FOUNDRY_HOSTNAME points to production enrollment
• [ ] Separate credentials from staging (not shared)
• [ ] Credential rotation schedule documented (90-day max)

Code Quality

• [ ] All tests passing including Foundry integration tests
• [ ] No hardcoded hostnames, tokens, or RIDs
• [ ] Error handling covers all Foundry ApiError status codes
• [ ] Rate limiting with exponential backoff implemented
• [ ] Logging uses structured format (JSON) with request IDs

Infrastructure

• [ ] Health check endpoint verifies Foundry connectivity

@app.get("/health")
async def health():
    try:
        client.ontologies.Ontology.list()
        return {"status": "healthy", "foundry": "connected"}
    except foundry.ApiError as e:
        return {"status": "degraded", "foundry": f"error_{e.status_code}"}

• [ ] Circuit breaker pattern for Foundry API calls
• [ ] Graceful degradation when Foundry is unreachable
• [ ] Timeout configuration: 30s for reads, 60s for writes
• [ ] Connection pooling configured

Monitoring & Alerting

• [ ] Metrics: request count, latency p50/p99, error rate by status code
• [ ] Alert: 5xx error rate > 5% for 5 minutes → P1
• [ ] Alert: p99 latency > 10s for 10 minutes → P2
• [ ] Alert: 429 rate > 10/min → P2 (tune rate limiter)
• [ ] Alert: 401/403 errors → P1 (credential issue)
• [ ] Dashboard with Foundry API health summary

Documentation

• [ ] Incident runbook: palantir-incident-runbook
• [ ] Credential rotation procedure documented
• [ ] Rollback procedure documented and tested
• [ ] On-call escalation path defined
• [ ] Foundry support contact info available

Deploy

set -euo pipefail
# Pre-flight
curl -sf "https://$FOUNDRY_HOSTNAME/api/v2/ontologies" \
  -H "Authorization: Bearer $FOUNDRY_TOKEN" > /dev/null \
  && echo "Foundry API reac…

Palantir Rate Limits

Overview

Handle Foundry API rate limits with exponential backoff, request queuing, and monitoring. Foundry rate limits vary by endpoint and enrollment tier.

Prerequisites

• foundry-platform-sdk installed
• Understanding of HTTP 429 responses

Instructions

Step 1: Understand Foundry Rate Limits

Foundry rate limits are per-user and per-endpoint. Key limits:

Rate limit headers returned:

• X-RateLimit-Limit — max requests per window
• X-RateLimit-Remaining — requests left in window
• Retry-After — seconds to wait (on 429)

Step 2: Implement Retry with Backoff (Python)

import time
import random
import foundry

def retry_foundry_call(fn, *args, max_retries=5, base_delay=1.0, **kwargs):
    """Retry Foundry API calls with jittered exponential backoff."""
    for attempt in range(max_retries + 1):
        try:
            return fn(*args, **kwargs)
        except foundry.ApiError as e:
            if attempt == max_retries:
                raise
            if e.status_code not in (429, 500, 502, 503):
                raise  # Non-retryable error
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            retry_after = getattr(e, "retry_after", None)
            if retry_after:
                delay = max(delay, float(retry_after))
            print(f"  Retry {attempt+1}/{max_retries} in {delay:.1f}s (HTTP {e.status_code})")
            time.sleep(delay)

# Usage
employees = retry_foundry_call(
    client.ontologies.OntologyObject.list,
    ontology="my-company", object_type="Employee", page_size=100,
)

Step 3: Request Queue for Batch Operations

import asyncio
from collections import deque

class FoundryRateLimiter:
    """Token bucket rate limiter for batch Foundry operations."""
    def __init__(self, max_per_second: int = 50):
        self.max_per_second = max_per_second
        self.tokens = max_per_second
        self._last_refill = time.monotonic()

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self._last_refill
        self.tokens = min(self.max_per_second, self.tokens + elapsed * self.max_per_second)
        self._last_refill = now

    def acquire(self):
        self._refill()
        if self.tokens < 1:…

Palantir Reference Architecture

Overview

Production-ready architecture for Foundry-integrated applications. Covers the standard data pipeline pattern (ingest > clean > model > serve), Ontology design, external API integration, and multi-repo project layout.

Prerequisites

• Foundry enrollment with project access
• Understanding of Ontology concepts (object types, link types, actions)
• Familiarity with palantir-core-workflow-a (transforms) and palantir-core-workflow-b (Ontology)

Instructions

Step 1: Data Pipeline Architecture

┌─────────────┐     ┌──────────────┐     ┌─────────────┐     ┌───────────┐
│  Raw Layer   │────>│  Clean Layer │────>│ Model Layer │────>│ Ontology  │
│ (ingested)   │     │  (validated) │     │ (enriched)  │     │ (objects) │
└─────────────┘     └──────────────┘     └─────────────┘     └───────────┘
  ↑ Connectors        @transform_df       @transform_df       Object types
  ↑ REST sync          null checks         joins, aggs         Link types
  ↑ File upload        type casting        ML features         Actions

Step 2: Project Layout (Foundry)

Foundry Project: "Customer Analytics"
├── Datasets/
│   ├── raw/                    # Ingested from sources
│   │   ├── raw_orders          # REST connector → CRM
│   │   ├── raw_customers       # JDBC connector → DB
│   │   └── raw_products        # File upload (CSV/Parquet)
│   ├── clean/                  # Validated, typed
│   │   ├── clean_orders        # Nulls removed, dates parsed
│   │   ├── clean_customers     # Deduped, normalized
│   │   └── clean_products      # Schema enforced
│   └── model/                  # Enriched, analytics-ready
│       ├── order_enriched      # Joined with customer + product
│       ├── customer_360        # Aggregated customer view
│       └── daily_summary       # Time-series aggregation
├── Code Repositories/
│   ├── pipeline-ingestion/     # Connectors and raw → clean
│   ├── pipeline-analytics/     # Clean → model transforms
│   └── ontology-actions/       # Action implementations
└── Ontology/
    ├── Object Types: Customer, Order, Product
    ├── Link Types: Customer→Orders, Order→Products
    └── Actions: createOrder, updateCustomerSegment

Step 3: External API Integration Pattern

# External app consuming Foundry Ontology via Platform SDK
my-external-app/
├── src/
│   ├── foundry/
│   │   ├── client.py           # Singleton FoundryClient
│   │   ├── objects.py          # Object query helpers
│   │   ├── actions.py          # Action wrappers
│   │   └── cache.py            # TTL cache layer
│   ├── api/
│   │   ├── routes.py           # REST endpoints
│   │   └── webhooks.py         # Foundry event handlers
│   └── main.py
├── tests/
│   ├── conftest.py             # Mocked FoundryClient
│   ├…

Palantir SDK Patterns

Overview

Production-ready patterns for Foundry Platform SDK and OSDK usage. Covers client singletons, typed error handling, pagination helpers, retry logic, and multi-tenant client factories.

Prerequisites

• Completed palantir-install-auth setup
• Familiarity with async/await patterns
• foundry-platform-sdk or @osdk/client installed

Instructions

Step 1: Singleton Client (Python)

# src/foundry_client.py
import os
import foundry
from functools import lru_cache

@lru_cache(maxsize=1)
def get_client() -> foundry.FoundryClient:
    """Thread-safe singleton — cached after first call."""
    auth = foundry.ConfidentialClientAuth(
        client_id=os.environ["FOUNDRY_CLIENT_ID"],
        client_secret=os.environ["FOUNDRY_CLIENT_SECRET"],
        hostname=os.environ["FOUNDRY_HOSTNAME"],
        scopes=["api:read-data", "api:write-data"],
    )
    auth.sign_in_as_service_user()
    return foundry.FoundryClient(auth=auth, hostname=os.environ["FOUNDRY_HOSTNAME"])

Step 2: Typed Error Handling

import foundry
from dataclasses import dataclass
from typing import TypeVar, Generic, Optional

T = TypeVar("T")

@dataclass
class Result(Generic[T]):
    data: Optional[T] = None
    error: Optional[str] = None
    status_code: Optional[int] = None

def safe_call(fn, *args, **kwargs) -> Result:
    """Wrap any Foundry SDK call with structured error handling."""
    try:
        return Result(data=fn(*args, **kwargs))
    except foundry.ApiError as e:
        return Result(error=e.message, status_code=e.status_code)
    except Exception as e:
        return Result(error=str(e))

# Usage
result = safe_call(
    get_client().ontologies.OntologyObject.list,
    ontology="my-company", object_type="Employee", page_size=10,
)
if result.error:
    print(f"Error {result.status_code}: {result.error}")
else:
    print(f"Found {len(result.data.data)} objects")

Step 3: Pagination Helper

def paginate_objects(client, ontology: str, object_type: str, page_size: int = 100):
    """Iterate through all objects with automatic pagination."""
    page_token = None
    while True:
        result = client.ontologies.OntologyObject.list(
            ontology=ontology,
            object_type=object_type,
            page_size=page_size,
            page_token=page_token,
        )
        yield from result.data
        page_token = result.next_page_token
        if not page_token:
            break

# Usage
for emp in paginate_objects(get_client(), "my-company", "Employee"):
   …

Palantir Security Basics

Overview

Security best practices for Foundry API tokens, OAuth2 credentials, scope management, and secret rotation. Covers both personal access tokens (dev) and service user credentials (production).

Prerequisites

• Foundry Developer Console access
• Understanding of OAuth2 scopes

Instructions

Step 1: Secure Credential Storage

# .env — NEVER commit to git
FOUNDRY_HOSTNAME=mycompany.palantirfoundry.com
FOUNDRY_CLIENT_ID=your-client-id
FOUNDRY_CLIENT_SECRET=your-client-secret

# .gitignore — ensure .env files are excluded
echo '.env' >> .gitignore
echo '.env.local' >> .gitignore
echo '.env.*.local' >> .gitignore

For production, use a secrets manager:

# AWS Secrets Manager
aws secretsmanager create-secret --name foundry/prod \
  --secret-string '{"client_id":"xxx","client_secret":"yyy","hostname":"zzz"}'

# Google Cloud Secret Manager
echo -n "your-client-secret" | gcloud secrets create foundry-client-secret --data-file=-

# HashiCorp Vault
vault kv put secret/foundry client_id=xxx client_secret=yyy

Step 2: Apply Least Privilege Scopes

# Production app that only reads Ontology objects:
auth = foundry.ConfidentialClientAuth(
    client_id=os.environ["FOUNDRY_CLIENT_ID"],
    client_secret=os.environ["FOUNDRY_CLIENT_SECRET"],
    hostname=os.environ["FOUNDRY_HOSTNAME"],
    scopes=["api:ontology-read"],  # Minimum viable scope
)

Step 3: Rotate Credentials

# 1. Generate new credentials in Developer Console
# 2. Deploy new credentials alongside old ones
# 3. Verify new credentials work
python -c "
import os, foundry
auth = foundry.ConfidentialClientAuth(
    client_id=os.environ['NEW_CLIENT_ID'],
    client_secret=os.environ['NEW_CLIENT_SECRET'],
    hostname=os.environ['FOUNDRY_HOSTNAME'],
    scopes=['api:read-data'],
)
auth.sign_in_as_service_user()
print('New credentials verified')
"
# 4. Remove old credentials from Developer Console
# 5. Update environment variables to use new credentials only

Step 4: Validate Tokens Are Not Exposed

…

Palantir Upgrade & Migration

Overview

Safely upgrade foundry-platform-sdk versions, handle breaking changes, and migrate between Foundry API versions. Includes a step-by-step upgrade checklist and rollback procedure.

Prerequisites

• Current foundry-platform-sdk installed
• Git for version control
• Test suite covering Foundry API calls
• Staging environment

Instructions

Step 1: Check Current Version and Available Updates

set -euo pipefail
pip show foundry-platform-sdk | grep -E "^(Name|Version)"
pip index versions foundry-platform-sdk 2>/dev/null | head -3
# Check OSDK version too
npm list @osdk/client 2>/dev/null || echo "OSDK not installed"

Step 2: Review Changelog

# Check GitHub releases for breaking changes
python -c "
import urllib.request, json
url = 'https://api.github.com/repos/palantir/foundry-platform-python/releases?per_page=5'
releases = json.loads(urllib.request.urlopen(url).read())
for r in releases:
    print(f'{r[\"tag_name\"]:12s} {r[\"published_at\"][:10]}')
    body = r.get('body', '')[:200]
    if 'BREAKING' in body.upper():
        print(f'  *** BREAKING CHANGES DETECTED ***')
    print()
"

Step 3: Create Upgrade Branch and Update

set -euo pipefail
git checkout -b upgrade/foundry-sdk-$(date +%Y%m%d)
pip install --upgrade foundry-platform-sdk
pip show foundry-platform-sdk | grep Version

Step 4: Run Tests and Fix Breaking Changes

set -euo pipefail
pytest tests/ -v --tb=short 2>&1 | tee upgrade-test-results.txt
# Review failures for breaking changes
grep -E "FAILED|ERROR" upgrade-test-results.txt

Common breaking changes between versions:

# v0.x → v1.x: Client initialization changed
# Before:
client = foundry.FoundryClient(auth=foundry.UserTokenAuth(token="..."))
# After:
client = foundry.FoundryClient(
    auth=foundry.UserTokenAuth(hostname="...", token="..."),
    hostname="...",
)

# v1.x → v2.x: Ontology methods moved
# Before:
client.ontology.list_objects(...)
# After:
client.ontologies.OntologyObject.list(...)

Step 5: Verify in Staging

# Deploy to staging and run smoke tests
FOUNDRY_HOSTNAME=$STAGING_HOSTNAME pytest tests/integration/ -v

Step 6: Rollback Procedure

# Pin previous version
pip install foundry-platform-sdk==0.8.0
# Or revert the branch
git checkout main -- requirements.txt
pip install -r requirements.txt

Output

• Updated SDK version with all tests passing
…

Palantir Webhooks & Events

Overview

Handle Foundry webhook events for Ontology changes, dataset updates, and build completions. Covers webhook registration via the Foundry API, signature verification, event routing, and idempotent processing.

Prerequisites

• Foundry enrollment with webhook support enabled
• HTTPS endpoint accessible from Foundry's network
• foundry-platform-sdk installed

Instructions

Step 1: Register a Webhook via API

import os, foundry

client = foundry.FoundryClient(
    auth=foundry.ConfidentialClientAuth(
        client_id=os.environ["FOUNDRY_CLIENT_ID"],
        client_secret=os.environ["FOUNDRY_CLIENT_SECRET"],
        hostname=os.environ["FOUNDRY_HOSTNAME"],
        scopes=["api:read-data", "api:write-data"],
    ),
    hostname=os.environ["FOUNDRY_HOSTNAME"],
)

# Register webhook for object change events
webhook = client.webhooks.Webhook.create(
    url="https://myapp.example.com/webhooks/foundry",
    event_types=["ontology.object.created", "ontology.object.updated"],
    secret="whsec_your_webhook_secret_here",
)
print(f"Webhook registered: {webhook.rid}")

Step 2: Webhook Endpoint with Signature Verification

from flask import Flask, request, jsonify
import hmac, hashlib

app = Flask(__name__)

@app.post("/webhooks/foundry")
def handle_foundry_webhook():
    # Verify signature
    signature = request.headers.get("X-Foundry-Signature", "")
    timestamp = request.headers.get("X-Foundry-Timestamp", "")
    secret = os.environ["FOUNDRY_WEBHOOK_SECRET"]

    signed_payload = f"{timestamp}.{request.get_data(as_text=True)}"
    expected = hmac.new(
        secret.encode(), signed_payload.encode(), hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({"error": "Invalid signature"}), 401

    # Replay protection — reject timestamps older than 5 minutes
    import time
    if abs(time.time() - int(timestamp)) > 300:
        return jsonify({"error": "Timestamp too old"}), 401

    event = request.get_json()
    handle_event(event)
    return jsonify({"received": True}), 200

Step 3: Event Router

def handle_event(event: dict):
    event_type = event.get("type", "")
    handlers = {
        "ontology.object.created": on_object_created,
        "ontology.object.updated": on_object_updated,
        "ontology.object.deleted": on_object_deleted,
        "dataset.updated": on_dataset_updated,
        "build.completed": on_build_com…