Flexport Sdk Patterns

Flexport SDK Patterns

Overview

Production-ready patterns for the Flexport REST API v2. Since Flexport has no official npm/pip SDK, you build typed HTTP clients. Key patterns: singleton client, paginated iteration, retry wrapper, and Zod response validation.

Prerequisites

• Scoped credentials from the secret manager, approved endpoint/destination policies, and fictional sandbox fixtures.
• An idempotency/retry design, redacted telemetry, and a clear owner for schema or permission failures.

Output

Return a client-validation receipt with contract version, fixture version, aggregate schema result, idempotency behavior, owner, and redacted failure reference. Do not log headers, documents, or shipment payloads.

Examples

Use a fictional shipment with an opaque ID to exercise pagination and a schema error. Verify that a retry cannot duplicate a write and that a rejected field is quarantined for review rather than forwarded to an external system.

Instructions

Pattern 1: Singleton Client with Auto-Retry

// src/flexport/client.ts
import { z } from 'zod';

class FlexportClient {
  private static instance: FlexportClient | null = null;
  private base = 'https://api.flexport.com';
  private headers: Record<string, string>;

  private constructor(apiKey: string) {
    this.headers = {
      'Authorization': `Bearer ${apiKey}`,
      'Flexport-Version': '2',
      'Content-Type': 'application/json',
    };
  }

  static getInstance(): FlexportClient {
    if (!this.instance) {
      const key = process.env.FLEXPORT_API_KEY;
      if (!key) throw new Error('Missing FLEXPORT_API_KEY');
      this.instance = new FlexportClient(key);
    }
    return this.instance;
  }

  async request<T>(path: string, options: RequestInit = {}): Promise<T> {
    const res = await fetch(`${this.base}${path}`, { ...options, headers: { ...this.headers, ...options.headers } });
    if (res.status === 429) {
      const retryAfter = parseInt(res.headers.get('Retry-After') || '60');
      await new Promise(r => setTimeout(r, retryAfter * 1000));
      return this.request(path, options);  // Retry once
    }
    if (!res.ok) {
      const body = await res.text();
      throw new FlexportAPIError(res.status, body, path);
    }
    return res.json();
  }
}

class FlexportAPIError extends Error {
  constructor(public status: number, public body: string, public path: string) {
    super(`Flexport ${status} on ${path}: ${body}`);
    this.name = 'FlexportAPIError';
  }
}

Pattern 2: Paginated Iterator

// Iterate all pages of a Flexport list endpoint
async function* paginate<T>(path: string, perPage = 25): AsyncGenerator<T> {
  const client = FlexportClient.getInstance();
  let page = 1;
  while (true) {
    const separator = path.includes('?') ? '&' : '?';
    const res = await client.request<{ data: { records: T[]; total_count: number } }>(
      `${path}${separator}page=${page}&per=${perPage}`
    );
    for (const record of res.data.records) yield record;
    if (res.data.records.length < perPage) break;
    page++;
  }
}

// Usage: iterate all shipments
for await (const shipment of paginate<Shipment>('/shipments')) {
  console.log(shipment.id, shipment.status);
}

Pattern 3: Zod Response Validation

const ShipmentSchema = z.object({
  id: z.string(),
  status: z.enum(['booked', 'in_transit', 'arrived', 'delivered']),
  freight_type: z.enum(['ocean', 'air', 'trucking']),
  origin_port: z.object({ code: z.string(), name: z.string() }),
  destination_port: z.object({ code: z.string(), name: z.string() }),
  cargo_ready_date: z.string(),
  estimated_arrival_date: z.string().nullable(),
});

type Shipment = z.infer<typeof ShipmentSchema>;

async function getShipment(id: string): Promise<Shipment> {
  const client = FlexportClient.getInstance();
  const res = await client.request<{ data: unknown }>(`/shipments/${id}`);
  return ShipmentSchema.parse(res.data);  // Throws ZodError on mismatch
}

Pattern 4: Python Typed Client

import os, requests
from dataclasses import dataclass
from typing import Iterator

@dataclass
class Shipment:
    id: str
    status: str
    freight_type: str

class FlexportClient:
    BASE = 'https://api.flexport.com'

    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {os.environ["FLEXPORT_API_KEY"]}',
            'Flexport-Version': '2',
        })

    def list_shipments(self, per: int = 25) -> Iterator[Shipment]:
        page = 1
        while True:
            r = self.session.get(f'{self.BASE}/shipments', params={'page': page, 'per': per})
            r.raise_for_status()
            records = r.json()['data']['records']
            for rec in records:
                yield Shipment(id=rec['id'], status=rec['status'], freight_type=rec['freight_type'])
            if len(records) < per:
                break
            page += 1

Error Handling

Resources

• Flexport API Reference
• Zod Documentation

Next Steps

Apply patterns in flexport-core-workflow-a for real-world usage.