pramiti-export-copilot

Push your semantic data model to dbt, Snowflake, Databricks, Looker, Tableau, GitHub, GitLab, or any HTTP ingestion endpoint whenever it changes. Zero dependencies -- inject any data backend and delivery target via Protocol-based interfaces.

Install

pip install pramiti-export-copilot

Quick Start

from pramiti_export_copilot import ExportCopilot, DeliveryConfig
from pramiti_export_copilot.backends import InMemoryBackend
from pramiti_export_copilot.delivery_targets import WebhookTarget

backend = InMemoryBackend(b"dbt schema content")
target = WebhookTarget("https://hooks.mycompany.com/epistom")
copilot = ExportCopilot(backend)
result = copilot.deliver("workspace-1", DeliveryConfig(format="dbt", target=target))

if result.success:
    print(f"Delivered {result.bytes_delivered} bytes, hash={result.export_hash[:8]}...")
elif result.error:
    print(f"Failed: {result.error}")

Supported Formats

Delivery Targets

All targets use urllib.request (stdlib) -- no external HTTP dependencies. All support self-hosted/enterprise instances via endpoint_url.

For GitHubTarget and GitLabTarget, the pre-flight existence check treats only HTTP 404 as "file absent" (create); any other pre-flight failure (auth, network, 5xx) is reported as a delivery error rather than silently creating a new file.

All targets return bytes_delivered as the size of the export content, not the transport envelope (JSON wrapper / base64 expansion).

Hash-Based Skip

Pass last_hash to skip delivery when content hasn't changed:

result = copilot.deliver("ws-1", config, last_hash="abc123...")
if result.skipped:
    print(f"Skipped: {result.reason}")  # "no_change"

API

Custom Backend

Implement IModelDataBackend (structural subtyping -- no inheritance required):

class MyBackend:
    def get_dbt_yaml(self, workspace_id: str) -> bytes: ...
    def get_snowflake_sql(self, workspace_id: str) -> bytes: ...
    def get_databricks_json(self, workspace_id: str) -> bytes: ...
    def get_lookml(self, workspace_id: str) -> bytes: ...
    def get_tableau_tds(self, workspace_id: str, connection_info: dict) -> bytes: ...

License

MIT -- Pramiti Labs