Client Architecture

┌─────────────────────────────────────────────────┐
│           Orchestrator (extractors/)            │
└──────────────┬──────────────────────────────────┘
               │
       ┌───────┴───────┐
       │               │
       v               v
┌─────────────┐  ┌──────────────┐
│  Confluent  │  │   Catalog    │
│   Clients   │  │  Providers   │
│             │  │              │
│ (extends    │  │ (implements  │
│  Confluent  │  │  Catalog     │
│  Client)    │  │  Provider)   │
└─────────────┘  └──────────────┘
       │               │
       v               v
┌─────────────┐  ┌──────────────┐
│  httpx      │  │ httpx/boto3/ │
│  AsyncClient│  │ google-auth  │
└─────────────┘  └──────────────┘

_RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
_MAX_RETRIES = 3
_BACKOFF_BASE = 1.0  # seconds

from lineage_bridge.clients.base import ConfluentClient

client = ConfluentClient(
    base_url="https://api.confluent.cloud",
    api_key="YOUR_API_KEY",
    api_secret="YOUR_API_SECRET",
)

async with client:
    # GET request
    data = await client.get("/org/v2/environments")

    # POST request
    result = await client.post("/path", json_body={"key": "value"})

    # Paginated request
    all_items = await client.paginate("/cmk/v2/clusters", params={"environment": "env-abc123"})

try:
    data = await client.get("/path")
except httpx.HTTPStatusError as e:
    if e.response.status_code == 401:
        print("Unauthorized - check API key")
    elif e.response.status_code == 403:
        print("Forbidden - check permissions")
    elif e.response.status_code == 404:
        print("Not found")

from typing import Protocol

class LineageExtractor(Protocol):
    async def extract(self) -> tuple[list[LineageNode], list[LineageEdge]]:
        """Extract lineage data and return nodes and edges."""
        ...

from lineage_bridge.clients.base import ConfluentClient
from lineage_bridge.clients.protocol import LineageExtractor
from lineage_bridge.models.graph import LineageNode, LineageEdge

class MyExtractor(ConfluentClient):
    async def extract(self) -> tuple[list[LineageNode], list[LineageEdge]]:
        nodes = []
        edges = []

        # Fetch data from API
        data = await self.get("/my/endpoint")

        # Build nodes and edges
        for item in data:
            node = LineageNode(...)
            nodes.append(node)

        return nodes, edges

from lineage_bridge.clients.kafka_admin import KafkaAdminClient

client = KafkaAdminClient(
    base_url="https://pkc-abc123.us-east-1.aws.confluent.cloud:443",
    api_key="KAFKA_KEY",
    api_secret="KAFKA_SECRET",
    cluster_id="lkc-xyz789",
    environment_id="env-abc123",
)

async with client:
    nodes, edges = await client.extract()

from lineage_bridge.clients.connect import ConnectClient

client = ConnectClient(
    api_key="CLOUD_API_KEY",
    api_secret="CLOUD_API_SECRET",
    environment_id="env-abc123",
    kafka_cluster_id="lkc-xyz789",
)

async with client:
    nodes, edges = await client.extract()

from lineage_bridge.clients.ksqldb import KsqlDBClient

client = KsqlDBClient(
    cloud_api_key="CLOUD_API_KEY",
    cloud_api_secret="CLOUD_API_SECRET",
    environment_id="env-abc123",
    ksqldb_api_key="KSQLDB_KEY",  # Optional
    ksqldb_api_secret="KSQLDB_SECRET",
)

async with client:
    nodes, edges = await client.extract()

from lineage_bridge.clients.flink import FlinkClient

client = FlinkClient(
    cloud_api_key="CLOUD_API_KEY",
    cloud_api_secret="CLOUD_API_SECRET",
    environment_id="env-abc123",
    organization_id="org-xyz",
    flink_api_key="FLINK_KEY",  # Optional
    flink_api_secret="FLINK_SECRET",
)

async with client:
    nodes, edges = await client.extract()

from lineage_bridge.clients.schema_registry import SchemaRegistryClient

client = SchemaRegistryClient(
    base_url="https://psrc-abc123.us-east-1.aws.confluent.cloud",
    api_key="SR_KEY",
    api_secret="SR_SECRET",
    environment_id="env-abc123",
)

async with client:
    nodes, edges = await client.extract()

from lineage_bridge.clients.stream_catalog import StreamCatalogClient

client = StreamCatalogClient(
    base_url="https://psrc-abc123.us-east-1.aws.confluent.cloud",
    api_key="SR_KEY",
    api_secret="SR_SECRET",
    environment_id="env-abc123",
)

async with client:
    await client.enrich(graph)  # Mutates graph

from lineage_bridge.clients.tableflow import TableflowClient

client = TableflowClient(
    api_key="CLOUD_API_KEY",
    api_secret="CLOUD_API_SECRET",
    environment_id="env-abc123",
    cluster_ids=["lkc-xyz789"],
)

async with client:
    nodes, edges = await client.extract()

from lineage_bridge.clients.metrics import MetricsClient

client = MetricsClient(
    api_key="CLOUD_API_KEY",
    api_secret="CLOUD_API_SECRET",
    lookback_hours=1,
)

async with client:
    enriched = await client.enrich(graph, cluster_id="lkc-xyz789")

from typing import Protocol

class CatalogProvider(Protocol):
    catalog_type: str  # "UNITY_CATALOG", "AWS_GLUE", "GOOGLE_DATA_LINEAGE"

    def build_node(
        self,
        catalog_name: str,
        schema_name: str,
        table_name: str,
        environment_id: str,
    ) -> tuple[LineageNode, LineageEdge]:
        """Create a catalog table node and MATERIALIZES edge."""
        ...

    async def enrich(self, graph: LineageGraph) -> None:
        """Backfill metadata from the catalog API."""
        ...

    def build_url(self, node: LineageNode) -> str:
        """Generate a deep link to the catalog UI."""
        ...

    async def push_lineage(
        self,
        graph: LineageGraph,
        *,
        on_progress: Callable[[str, str], None] | None = None,
    ) -> PushResult:
        """Write lineage metadata back to the catalog."""
        ...

from lineage_bridge.catalogs.databricks_uc import DatabricksUCProvider

provider = DatabricksUCProvider(
    workspace_url="https://myworkspace.cloud.databricks.com",
    token="dapi...",
)

await provider.enrich(graph)

from lineage_bridge.catalogs.aws_glue import GlueCatalogProvider

provider = GlueCatalogProvider(region="us-east-1")

await provider.enrich(graph)

from lineage_bridge.catalogs.google_lineage import GoogleLineageProvider

provider = GoogleLineageProvider(
    project_id="my-project",
    location="us-central1",
)

await provider.enrich(graph)

_EXTRACTOR_TIMEOUT = 120  # seconds

_DEFAULT_TIMEOUT = 30.0  # seconds

_MAX_RETRIES = 3
_BACKOFF_BASE = 1.0  # seconds (exponential: 1s, 2s, 4s)