> ## Documentation Index
> Fetch the complete documentation index at: https://docs.gable.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Guide: Auto-publish Data Store Schemas from Avro and Protobuf

> Automatically publish data store schema components whenever schema files change in your repository

Use this guide to automatically publish data store schema components from `.avsc` and `.proto` files whenever changes are pushed to your repository.

This workflow scans for schema files, extracts and combines their fields using the `recap-core` library, and publishes to Gable's API as a `DATA_STORE` component.

For managing data store components in the UI, see [Services & Data Components](/docs/onboarding/services-and-data-components).

## 1. Set up your repo

This guide assumes the following repo structure. Add your schema files, publishing script, and GitHub Action using this layout (or adapt the guide to fit your repo structure if different).

```
your-repo/
├── schemas/
│   ├── avro/*.avsc
│   └── proto/*.proto
├── scripts/publish_schemas.py
└── .github/workflows/publish-schemas.yml
```

Install the required Python dependencies:

```
requests
# Avro's dependencies are part of the base recap-core package
recap-core[proto]
```

## 2. Add the scan + publish script

Create `scripts/publish_schemas.py`.

The script:

1. Starts a run to track the relevant `git` info
2. Finds all `.avsc` and `.proto` files under `schemas/`.
3. Converts each file to Recap schema format.
4. Combines their fields into a single schema.
5. Publishes the `DATA_STORE` component to Gable's ingestion API.

```python theme={null}
import os
import pathlib

import requests
from recap.converters.avro import AvroConverter
from recap.converters.protobuf import ProtobufConverter
from recap.types import to_dict

GABLE_API_ENDPOINT = os.environ["GABLE_API_ENDPOINT"]
GABLE_API_KEY = os.environ["GABLE_API_KEY"]
COMPONENT_ID = os.environ["COMPONENT_ID"]
TABLE_ID = os.environ["TABLE_ID"]
TABLE_NAME = os.environ["TABLE_NAME"]
ROOT = pathlib.Path(os.environ.get("SCHEMA_ROOT", "schemas"))

def extract_fields(path: pathlib.Path):
    '''
    Uses Gable's recap-core library to parse Avro and Proto files
    and return their fields
    '''
    text = path.read_text()
    if path.suffix == ".avsc":
        recap_type = AvroConverter().to_recap(text)
    elif path.suffix == ".proto":
        recap_type = ProtobufConverter().to_recap(text)
    else:
        return []
    return to_dict(recap_type)["fields"]

def start_run():
    start_run_request = {
        "action": "upload",
        "type": "DATA_STORE",
        "collection_mechanism": "BYOL",
        "code_info": {
            "namespace": "qa",
            "repo_uri": f"{os.environ['GITHUB_SERVER_URL']}/{os.environ['GITHUB_REPOSITORY']}",
            "repo_branch": os.environ["GITHUB_REF_NAME"],
            "repo_commit": os.environ["GITHUB_SHA"],
            "project_root": "/",
            "repo_name": os.environ["GITHUB_REPOSITORY"].split("/")[-1],
            "job_trigger": "MANUAL",
            "external_component_id": COMPONENT_ID,
        },
    }
    r = requests.post(
        f"{GABLE_API_ENDPOINT}/v0/sca/start-run",
        json=start_run_request,
        headers={"x-api-key": GABLE_API_KEY},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()["runId"]

def publish(payload):
    r = requests.post(
        f"{GABLE_API_ENDPOINT}/v0/sca/results",
        json=payload,
        headers={"x-api-key": GABLE_API_KEY},
        timeout=30,
    )
    r.raise_for_status()

def main():
    run_id = start_run()
    fields = []
    # Parse each file using recap and combine their schemas
    for path in sorted(ROOT.rglob("*")):
        if path.suffix not in (".avsc", ".proto"):
            continue
        fields.extend(extract_fields(path))
        print(f"collected {path}")
    # Construct the component's payload
    payload = {
        "external_component_id": COMPONENT_ID,
        "type": "DATA_STORE",
        "run_id": run_id,
        "external_table_id": TABLE_ID,
        "table_metadata": {
            "type": "dynamodb",
            "table_name": TABLE_NAME,
        },
        "schema": {"fields": fields},
    }
    publish(payload)

if __name__ == "__main__":
    main()
```

## 3. Configure the API payload

The API expects one POST per schema with this shape:

```json theme={null}
{
  "external_component_id": "DA32FCF6-C49D-4329-9033-EC298480930D",
  "run_id": "7C5CB5DB-B436-4E42-83BB-84512BA56369",
  "type": "DATA_STORE",
  "external_table_id": "7A253847-CF7B-470E-8EB4-CAE6DFD2F70B",
  "table_metadata": {
    "type": "dynamodb",
    "table_name": "dynamo table"
  },
  "schema": {
    "fields": [
      { "name": "field_1", "type": "string" }
    ]
  }
}
```

* `external_component_id` is the UUID that the data store component will be stored under in Gable. It can be any stable UUID that uniquely identifies the data store component.
* `run_id` identifies the run associated with the lineage version.
* `external_table_id` identifies the specific schema or table.

## 4. Add the GitHub Action

Create `.github/workflows/publish-schemas.yml`:

```yaml theme={null}
name: Publish schemas
on:
  push:
    branches: [main]
    # The path filter prevents this workflow from running
    # if the schema files haven't changed!
    paths:
      - "schemas/**/*.avsc"
      - "schemas/**/*.proto"
  workflow_dispatch:

jobs:
  publish:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -r requirements.txt
      - run: python scripts/publish_schemas.py
        env:
          GABLE_API_ENDPOINT: ${{ vars.GABLE_API_ENDPOINT }}
          GABLE_API_KEY: ${{ secrets.GABLE_API_KEY }}
```

## 5. Trigger publish of schemas

To publish schemas automatically, merge changes to `.avsc` or `.proto` files into `main`.

After the workflow runs, confirm that the data store schemas are visible in Gable and that the generated data store component reflects the expected fields.