Connector SDK¶

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class RawEvent:
    """A raw audit event from a source platform."""
    platform: str               # e.g. "azure_ad", "fabric", "databricks"
    event_id: str               # Unique event ID from the source
    occurred_at: datetime       # When the event occurred (UTC)
    principal_ref: str          # Source-specific user/principal reference
    asset_ref: str              # Source-specific asset/resource reference
    operation: str              # e.g. "SELECT", "INSERT", "LOGIN"
    raw_payload: dict = field(default_factory=dict)  # Full original payload

@dataclass
class PermissionSnapshot:
    """A point-in-time snapshot of a permission grant."""
    platform: str               # Source platform identifier
    principal_ref: str          # Source-specific user reference
    asset_ref: str              # Source-specific asset reference
    privilege: str              # e.g. "READ", "WRITE", "ADMIN"
    grant_mechanism: str        # e.g. "direct", "group", "role", "policy"
    granted_via: str | None = None      # Group/role name, if applicable
    snapshot_at: datetime | None = None # When the snapshot was taken
    raw_payload: dict = field(default_factory=dict)

class BaseConnector(ABC):
    def __init__(self, config: Any, kafka_producer: Any, db_pool: Any) -> None:
        self.config = config
        self.kafka_producer = kafka_producer
        self.db_pool = db_pool

@abstractmethod
async def stream_audit_events(self, since: datetime) -> AsyncIterator[RawEvent]:
    ...

@abstractmethod
async def snapshot_permissions(self) -> AsyncIterator[PermissionSnapshot]:
    ...

async def run(self) -> None:
    checkpoint = await self._load_checkpoint()
    async for event in self.stream_audit_events(since=checkpoint):
        await self.kafka_producer.send(
            topic=f"verity.events.raw.{event.platform}",
            value={ ... },  # Serialised RawEvent fields
        )
        await self._save_checkpoint(event.occurred_at)

"""Connector for Acme Data Platform."""

from __future__ import annotations

import logging
from datetime import datetime
from typing import Any, AsyncIterator

import httpx

from verity_sdk.connector_base import BaseConnector, PermissionSnapshot, RawEvent

logger = logging.getLogger(__name__)


class AcmeConnector(BaseConnector):
    """Ingests audit events and permissions from the Acme Data Platform."""

    PLATFORM = "acme"

    def __init__(self, config: Any, kafka_producer: Any, db_pool: Any) -> None:
        super().__init__(config, kafka_producer, db_pool)
        self.base_url = config["acme_api_url"]
        self.api_key = config["acme_api_key"]
        self.client = httpx.AsyncClient(
            base_url=self.base_url,
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=30.0,
        )

    async def stream_audit_events(self, since: datetime) -> AsyncIterator[RawEvent]:
        """Poll the Acme audit log API for new events."""
        page = 1
        while True:
            resp = await self.client.get(
                "/api/v1/audit-logs",
                params={
                    "since": since.isoformat(),
                    "page": page,
                    "page_size": 100,
                },
            )
            resp.raise_for_status()
            data = resp.json()

            for entry in data["events"]:
                yield RawEvent(
                    platform=self.PLATFORM,
                    event_id=entry["id"],
                    occurred_at=datetime.fromisoformat(entry["timestamp"]),
                    principal_ref=entry["user_email"],
                    asset_ref=entry["resource_urn"],
                    operation=entry["action"],
                    raw_payload=entry,
                )

            if not data.get("has_next"):
                break
            page += 1

    async def snapshot_permissions(self) -> AsyncIterator[PermissionSnapshot]:
        """Fetch all current permission grants from Acme."""
        resp = await self.client.get("/api/v1/permissions")
        resp.raise_for_status()

        for grant in resp.json()["grants"]:
            yield PermissionSnapshot(
                platform=self.PLATFORM,
                principal_ref=grant["user_email"],
                asset_ref=grant["resource_urn"],
                privilege=grant["level"].upper(),
                grant_mechanism=grant.get("mechanism", "direct"),
                granted_via=grant.get("role_name"),
                snapshot_at=datetime.utcnow(),
                raw_payload=grant,
            )


# ── Entry point ──────────────────────────────────────────────────────────
async def main():
    from verity_common.config import VerityConfig
    from verity_common.kafka import KafkaSettings, create_producer

    config = VerityConfig()
    connector_config = {
        "connector_id": "acme-prod",
        "acme_api_url": "https://api.acme.example.com",
        "acme_api_key": "sk-...",
    }

    async with create_producer(config.kafka) as producer:
        import asyncpg
        pool = await asyncpg.create_pool(dsn="postgresql://verity:verity@localhost/verity")
        connector = AcmeConnector(connector_config, producer, pool)
        await connector.run()


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())