Building an Integration - Coalesce Quality Documentation

A custom integration is a program that reads your tool and writes what it finds to the Coalesce Quality API. It is idempotent — running it twice produces the same estate — so it is a cron job, not a migration. Every snippet below is taken from the runnable examples in getsynq/api/examples. Pick a language in any code block and the rest of the page follows.

The complete program these come from is bi_tool_lineage (Python) — a whole BI tool modelled end to end, run against a live workspace.

Credentials

Create client credentials in API settings — /settings/api in your own app, which is app.synq.io for EU, app.us.synq.io for US and app.au.synq.io for AU. Grant these scopes:

Scope	Needed for
Edit Entity Types	Step 1
Edit Entities	Steps 2, 3, 4, 6, 7
Edit Lineage	Step 5
Edit Executions	Step 8
Read Entities, Read Entity Types, Read Lineage	Steps 1 and 9

Getting Started covers exchanging them for an access token and generating or installing a client — follow it rather than the short version, it is generated from the API itself and stays current. Your region’s API endpoint is in the region table. What is specific to this guide is the environment it expects and the gRPC channel:

export QUALITY_CLIENT_ID=...
export QUALITY_CLIENT_SECRET=...
export QUALITY_API_ENDPOINT=...   # your region's API endpoint

go get buf.build/gen/go/getsynq/api/grpc/go
go get buf.build/gen/go/getsynq/api/protocolbuffers/go

export QUALITY_CLIENT_ID=...
export QUALITY_CLIENT_SECRET=...
export QUALITY_API_ENDPOINT=...   # your region's API endpoint

pip install grpcio protobuf requests
pip install getsynq-api-protocolbuffers-python getsynq-api-grpc-python \
  --extra-index-url https://buf.build/gen/python

Send the token as authorization: Bearer <token> on every call and refresh it when expires_in elapses. Set the gRPC authority explicitly — without it the port travels in the Host header and every RPC returns UNIMPLEMENTED with no message, which reads as a missing service rather than a routing miss:

import (
    "golang.org/x/oauth2"
    "golang.org/x/oauth2/clientcredentials"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/credentials/oauth"
)

cfg := clientcredentials.Config{
    ClientID:     os.Getenv("QUALITY_CLIENT_ID"),
    ClientSecret: os.Getenv("QUALITY_CLIENT_SECRET"),
    TokenURL:     "https://" + endpoint + "/oauth2/token",
}
conn, err := grpc.NewClient(endpoint+":443",
    grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{})),
    grpc.WithPerRPCCredentials(oauth.TokenSource{
        TokenSource: cfg.TokenSource(ctx),
    }),
    // Required. Without it the port travels in the Host header and every RPC
    // comes back UNIMPLEMENTED with no message.
    grpc.WithAuthority(endpoint),
)

import grpc, requests, time

class TokenSource:
    def __init__(self, client_id, client_secret, api_endpoint):
        self.token_url = f"https://{api_endpoint}/oauth2/token"
        self.client_id, self.client_secret = client_id, client_secret
        self.obtain_token()

    def obtain_token(self):
        resp = requests.post(self.token_url, data={
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "grant_type": "client_credentials",
        })
        resp.raise_for_status()
        self.token = resp.json()
        self.expires_at = time.time() + self.token["expires_in"]

    def get_token(self) -> str:
        if time.time() > self.expires_at:
            self.obtain_token()
        return self.token["access_token"]

class TokenAuth(grpc.AuthMetadataPlugin):
    def __init__(self, source): self._source = source
    def __call__(self, context, callback):
        callback([("authorization", f"Bearer {self._source.get_token()}")], None)

channel = grpc.secure_channel(
    f"{endpoint}:443",
    grpc.composite_channel_credentials(
        grpc.ssl_channel_credentials(),
        grpc.metadata_call_credentials(TokenAuth(TokenSource(cid, secret, endpoint))),
    ),
    # Required. Without an explicit authority the port travels in the Host
    # header and every RPC comes back UNIMPLEMENTED with no message.
    options=(("grpc.default_authority", endpoint),),
)

1. Types

One UpsertType per kind of thing your tool has. Declare traits — they decide how your entities rank, sort and behave platform-wide.

types := []*entitiesv1.Type{
    {
        TypeId:  31,
        Name:    "Metabase Model",
        SvgIcon: iconLayers,
        Traits:  &entitiesv1.TypeTraits{IsModel: proto.Bool(true)},
    },
    {
        TypeId:  32,
        Name:    "Metabase Question",
        SvgIcon: iconChart,
        // A question transforms data AND is read by a person, so it is both.
        Traits: &entitiesv1.TypeTraits{IsModel: proto.Bool(true), IsBiLike: proto.Bool(true)},
    },
    {
        TypeId:  34,
        Name:    "Metabase Alert",
        SvgIcon: iconWarning,
        // A check, not a stage data flows through.
        Traits: &entitiesv1.TypeTraits{IsTestType: proto.Bool(true)},
    },
}

for _, t := range types {
    if _, err := api.types.UpsertType(ctx, &customv1.UpsertTypeRequest{Type: t}); err != nil {
        return err
    }
}

types = [
    type_pb2.Type(
        type_id=31,
        name="Metabase Model",
        svg_icon=ICON_LAYERS,
        traits=type_traits_pb2.TypeTraits(is_model=True),
    ),
    type_pb2.Type(
        type_id=32,
        name="Metabase Question",
        svg_icon=ICON_CHART,
        # A question transforms data AND is read by a person, so it is both.
        traits=type_traits_pb2.TypeTraits(is_model=True, is_bi_like=True),
    ),
    type_pb2.Type(
        type_id=34,
        name="Metabase Alert",
        svg_icon=ICON_WARNING,
        # A check, not a stage data flows through.
        traits=type_traits_pb2.TypeTraits(is_test_type=True),
    ),
]

for t in types:
    api.types.UpsertType(types_service_pb2.UpsertTypeRequest(type=t))

Type numbers are 1–1000 and workspace-wide, with no allocator. Call ListTypes first and refuse to start if one of your numbers is taken under a different name — that is the only protection against two integrations picking the same band.

2. Entities

One UpsertEntity per object. Create every entity before anything points at one: a SQL binding to an entity that does not exist is rejected at the write, and an entity group whose members do not all exist is not a set anyone can reconcile against.

entity := &entitiesv1.Entity{
    Id:          customID("metabase::question::87"),
    TypeId:      32,
    Name:        "Revenue by product category",
    Description: "Revenue split by category, last 90 days.",
    Annotations: []*entitiesv1.Annotation{
        {Name: "metabase.collection", Values: []string{"Commercial"}},
        {Name: "metabase.kind", Values: []string{"question"}},
    },
}
_, err := api.entities.UpsertEntity(ctx, &customv1.UpsertEntityRequest{Entity: entity})

entity = entity_pb2.Entity(
    id=custom_id("metabase::question::87"),
    type_id=32,
    name="Revenue by product category",
    description="Revenue split by category, last 90 days.",
    annotations=[
        annotation_pb2.Annotation(name="metabase.collection", values=["Commercial"]),
        annotation_pb2.Annotation(name="metabase.kind", values=["question"]),
    ],
)
api.entities.UpsertEntity(entities_service_pb2.UpsertEntityRequest(entity=entity))

Where customID builds the identifier:

func customID(id string) *entitiesv1.Identifier {
    return &entitiesv1.Identifier{
        Id: &entitiesv1.Identifier_Custom{
            Custom: &entitiesv1.CustomIdentifier{Id: id},
        },
    }
}

def custom_id(id_: str) -> identifier_pb2.Identifier:
    return identifier_pb2.Identifier(custom=identifier_pb2.CustomIdentifier(id=id_))

The id is a slug; the readable text belongs in name. Keep it to a lowercase A-Z a-z 0-9 : _ . - slug built from the source system’s stable id. Anything else is folded on write — metabase/question/87 and metabase question 87 are the same entity — and ids are case-sensitive, so a capital makes a different one. name takes 500 characters of any UTF-8, which is where the title, the accents and the spaces go. See Entity ids.

To point at an entity from a connected platform, use that platform’s own identifier variant rather than custom:

&entitiesv1.Identifier{
    Id: &entitiesv1.Identifier_BigqueryTable{
        BigqueryTable: &entitiesv1.BigqueryTableIdentifier{
            Project: cfg.project, Dataset: cfg.dataset, Table: "order_items",
        },
    },
}

identifier_pb2.Identifier(
    bigquery_table=identifier_pb2.BigqueryTableIdentifier(
        project=cfg.project, dataset=cfg.dataset, table="order_items",
    )
)

Where the platform has no structured variant, or you already hold the entity’s opaque entity_id, use the synq_path variant instead — see Finding the entity_id. Annotations are your tool’s own filing system carried across, so the catalog can be filtered the way the tool is browsed. Keep them to labels a person would pick out of a list; prose belongs in the description, which accepts Markdown.

3. Schemas

Declare columns before any lineage step. A SELECT * expands only over an entity whose columns are known, and a declared column edge only shows against a column the entity has. Skip this and the table-level edges still appear, with nothing saying the column ones are missing.

schema := &customfeaturesv1.Schema{StateAt: timestamppb.Now()}
for i, c := range columns {
    schema.Columns = append(schema.Columns, &entitiesv1.SchemaColumn{
        Name:            c.name,
        NativeType:      c.nativeType,
        Description:     c.desc,
        OrdinalPosition: int32(i + 1),
    })
}

_, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{
    Feature: &customv1.Feature{
        EntityId: customID("metabase::question::87"),
        // One schema feature per entity, so the id is a constant. Never generate
        // it — a fresh id on every run creates a new feature each time instead
        // of replacing the previous one.
        FeatureId: "schema",
        Feature:   &customv1.Feature_Schema{Schema: schema},
    },
})

schema = schema_pb2.Schema(
    state_at=now(),
    columns=[
        entities_schema_pb2.SchemaColumn(
            name=c.name,
            native_type=c.native_type,
            description=c.desc,
            ordinal_position=i + 1,
        )
        for i, c in enumerate(columns)
    ],
)

api.features.UpsertEntityFeature(
    features_service_pb2.UpsertEntityFeatureRequest(
        feature=features_service_pb2.Feature(
            entity_id=custom_id("metabase::question::87"),
            # One schema feature per entity, so the id is a constant. Never
            # generate it — a fresh id on every run creates a new feature each
            # time instead of replacing the previous one.
            feature_id="schema",
            schema=schema,
        )
    )
)

4. Lineage features

SqlDefinition, with bindings

The mechanism to reach for first. Give the SQL as the tool executes it, plus the dialect and a database_context for unqualified names. A name in the SQL that is not a warehouse object — another custom entity, a queue, a service — resolves to nothing and is silently dropped. references binds such a name to the entity it stands for. Names that are warehouse objects need no binding.

_, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{
    Feature: &customv1.Feature{
        EntityId:  customID("metabase::question::87"),
        FeatureId: "sql",
        Feature: &customv1.Feature_SqlDefinition{
            SqlDefinition: &customfeaturesv1.SqlDefinition{
                StateAt:         timestamppb.Now(),
                Dialect:         entitiesv1.SqlDialect_SQL_DIALECT_BIGQUERY,
                Sql:             question.sql,
                DatabaseContext: warehouseContext(cfg),
                References: []*customfeaturesv1.SqlTableReference{{
                    // The name exactly as the SQL writes it. Matched
                    // case-insensitively; leave database_name and schema_name
                    // empty to default them from database_context.
                    ObjectName: "order_items_enriched",
                    Entity:     customID("metabase::model::31"),
                }},
            },
        },
    },
})

api.features.UpsertEntityFeature(
    features_service_pb2.UpsertEntityFeatureRequest(
        feature=features_service_pb2.Feature(
            entity_id=custom_id("metabase::question::87"),
            feature_id="sql",
            sql_definition=sql_definition_pb2.SqlDefinition(
                state_at=now(),
                dialect=sql_dialect_pb2.SQL_DIALECT_BIGQUERY,
                sql=question.sql,
                database_context=warehouse_context(cfg),
                references=[
                    sql_definition_pb2.SqlTableReference(
                        # The name exactly as the SQL writes it. Matched
                        # case-insensitively; leave database_name and schema_name
                        # empty to default them from database_context.
                        object_name="order_items_enriched",
                        entity=custom_id("metabase::model::31"),
                    )
                ],
            ),
        )
    )
)

The bindings are part of the definition and replace with it: one dropped from a later write is gone, and the lineage it produced is withdrawn.

ColumnLineage

For a component with no SQL at all. State each edge outright. The declaration is complete and replaces the previous one whole, so regenerate it from your tool’s metadata every run rather than diffing.

var edges []*customfeaturesv1.ColumnEdge
for _, card := range dashboard.cards {
    edges = append(edges, &customfeaturesv1.ColumnEdge{
        Upstream:       customID(fmt.Sprintf("metabase::question::%d", card.fromQuestion)),
        UpstreamColumn: card.fromColumn,
        // The two sides are named independently, so they need not match.
        Column: card.field,
    })
}

_, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{
    Feature: &customv1.Feature{
        EntityId: customID("metabase::dashboard::9"),
        // Only one column-lineage feature per entity, so a stable id is how it
        // is edited.
        FeatureId: "column-lineage",
        Feature: &customv1.Feature_ColumnLineage{
            ColumnLineage: &customfeaturesv1.ColumnLineage{
                StateAt: timestamppb.Now(),
                Edges:   edges,
            },
        },
    },
})

edges = [
    column_lineage_pb2.ColumnEdge(
        upstream=custom_id(f"metabase::question::{c.from_question}"),
        upstream_column=c.from_column,
        # The two sides are named independently, so they need not match.
        column=c.field,
    )
    for c in dashboard.cards
]

api.features.UpsertEntityFeature(
    features_service_pb2.UpsertEntityFeatureRequest(
        feature=features_service_pb2.Feature(
            entity_id=custom_id("metabase::dashboard::9"),
            # Only one column-lineage feature per entity, so a stable id is how
            # it is edited.
            feature_id="column-lineage",
            column_lineage=column_lineage_pb2.ColumnLineage(state_at=now(), edges=edges),
        )
    )
)

An upstream on a connected platform works the same way, and does not have to exist yet — the edge is remembered and appears once the entity is ingested. Declaring a column edge also declares the table edge it implies.

Code and Git

Code carries the SQL, Python, YAML or JSON behind the entity for display; several per entity. GitFileReference says where that code lives, so a connected code integration can show recent commits against the entity when it fails.

5. Relationships

For edges with nothing to say at column grain. Read the response — an edge is held between the two entities the endpoints resolve to, so two relationships that look different can be the same edge, and only the last write survives.

resp, err := api.relationships.UpsertRelationships(ctx, &customv1.UpsertRelationshipsRequest{
    Relationships: []*customv1.Relationship{{
        Upstream:   customID("metabase::dashboard::9"),
        Downstream: customID("metabase::subscription::4"),
    }},
})
if err != nil {
    return err
}
for _, r := range resp.GetResults() {
    fmt.Printf("%s -> %s  %s\n",
        r.GetRelationship().GetUpstream().GetEntityId(),
        r.GetRelationship().GetDownstream().GetEntityId(),
        r.GetOutcome(),
    )
}

resp = api.relationships.UpsertRelationships(
    relationships_service_pb2.UpsertRelationshipsRequest(
        relationships=[
            relationships_service_pb2.Relationship(
                upstream=custom_id("metabase::dashboard::9"),
                downstream=custom_id("metabase::subscription::4"),
            )
        ]
    )
)
for r in resp.results:
    outcome = relationships_service_pb2.RelationshipWriteOutcome.Name(r.outcome)
    print(f"{r.relationship.upstream.entity_id} -> "
          f"{r.relationship.downstream.entity_id}  {outcome}")

6. Checks

Something that validates another entity is not a stage data flows through. Attach it with a check relationship, then give it a CheckCategory feature — attaching schedules no re-publish of the derived check record, while writing a feature does, so the reverse order leaves that record stale until the entity is written again.

_, err := api.features.UpsertEntityFeature(ctx, &customv1.UpsertEntityFeatureRequest{
    Feature: &customv1.Feature{
        EntityId:  customID("metabase::alert::12"),
        FeatureId: "check",
        Feature: &customv1.Feature_CheckCategory{
            // category and governance_category are left unset on purpose: a
            // producer value outranks the workspace's categorisation rules.
            CheckCategory: &customfeaturesv1.CheckCategory{
                Package: "metabase",
                Kind:    alert.kind,
            },
        },
    },
})

_, err = api.checkRelationships.UpsertCheckRelationships(ctx,
    &customv1.UpsertCheckRelationshipsRequest{
        CheckRelationships: []*customv1.CheckRelationship{{
            Check:          customID("metabase::alert::12"),
            Checked:        customID("metabase::question::87"),
            CheckedColumns: []string{"revenue"},
        }},
    })

api.features.UpsertEntityFeature(
    features_service_pb2.UpsertEntityFeatureRequest(
        feature=features_service_pb2.Feature(
            entity_id=custom_id("metabase::alert::12"),
            feature_id="check",
            # category and governance_category are left unset on purpose: a
            # producer value outranks the workspace's categorisation rules.
            check_category=checks_pb2.CheckCategory(package="metabase", kind=alert.kind),
        )
    )
)

api.check_relationships.UpsertCheckRelationships(
    checks_relationships_service_pb2.UpsertCheckRelationshipsRequest(
        check_relationships=[
            checks_relationships_service_pb2.CheckRelationship(
                check=custom_id("metabase::alert::12"),
                checked=custom_id("metabase::question::87"),
                checked_columns=["revenue"],
            )
        ]
    )
)

Leave category and governance_category unset. They are resolved by your workspace’s own categorisation rules, and a value sent by a producer outranks those rules. Deriving one from your tool’s alert kind — the obvious thing to do — silently switches off every rule the workspace wrote, and nothing in the product explains why.

7. The entity group

One group naming every entity your integration owns, every run. The server diffs it against the previous set and deletes what is missing, so an object deleted in your tool disappears from the catalog with no state kept on your side.

ids := everyEntity() // the FULL set, every run
resp, err := api.groups.UpsertEntitiesGroup(ctx, &customv1.UpsertEntitiesGroupRequest{
    Group: &customv1.Group{GroupId: "metabase", EntityIds: ids},
})
if err != nil {
    return err
}
for _, deleted := range resp.GetDeletedIds() {
    fmt.Printf("deleted (gone from the tool): %s\n", deleted.GetCustom().GetId())
}

ids = every_entity()  # the FULL set, every run
resp = api.groups.UpsertEntitiesGroup(
    groups_service_pb2.UpsertEntitiesGroupRequest(
        group=groups_service_pb2.Group(group_id="metabase", entity_ids=ids)
    )
)
for deleted in resp.deleted_ids:
    print(f"deleted (gone from the tool): {deleted.custom.id}")

A partial send is a deletion of everything omitted. If reading your tool can fail halfway, skip the group write for that run and let the next complete one reconcile.

A group deletes entities only — not the features, relationships or check relationships of entities that still exist. A repointed dashboard, a removed check or a stale SqlDefinition survives the group write and stays in the graph.

So each run also reconciles what the last one wrote: read the current state with ListEntityFeatures (leave entity_id unset for the whole workspace), ListRelationships and ListCheckRelationships; diff it against what you are about to write; then call DeleteEntityFeature, DeleteRelationships and DeleteCheckRelationships for what is gone — before the group write, so no edge is left pointing at an entity that is about to go.

8. Executions

What gives a custom entity a status rather than just a position in a graph. created_at is required and must be in the past, so a run that has not happened yet is simply not reported.

_, err := api.executions.UpsertExecution(ctx, &customv1.UpsertExecutionRequest{
    Execution: &customv1.Execution{
        Id:         customID("metabase::dashboard::9"),
        Status:     customv1.ExecutionStatus_EXECUTION_STATUS_OK,
        Message:    "All cards refreshed.",
        CreatedAt:  timestamppb.New(finished),
        StartedAt:  timestamppb.New(started),
        FinishedAt: timestamppb.New(finished),
    },
})

api.executions.UpsertExecution(
    entity_executions_service_pb2.UpsertExecutionRequest(
        execution=entity_executions_service_pb2.Execution(
            id=custom_id("metabase::dashboard::9"),
            status=entity_executions_service_pb2.EXECUTION_STATUS_OK,
            message="All cards refreshed.",
            created_at=finished,
            started_at=started,
            finished_at=finished,
        )
    )
)

Status is OK, WARN, ERROR or CRITICAL. UpsertLogEntry attaches log output without changing status.

9. Read it back

Every way a custom integration goes wrong produces a valid-looking write and an empty graph, and the write side cannot tell the difference. End the program by reading its own lineage back. Lineage is computed asynchronously, so poll.

depth := int32(10)
resp, err := api.lineage.GetLineage(ctx, &lineagev1.GetLineageRequest{
    LineageDirection: lineagev1.LineageDirection_LINEAGE_DIRECTION_UPSTREAM,
    StartPoint: &lineagev1.GetLineageStartPoint{
        From: &lineagev1.GetLineageStartPoint_Entities{
            Entities: &lineagev1.EntitiesStartPoint{
                Entities: []*entitiesv1.Identifier{customID("metabase::dashboard::9")},
            },
        },
    },
    MaxDepth: &depth,
})
if err != nil {
    return err
}
lin := resp.GetLineage()
if len(lin.GetNodeDependencies()) == 0 {
    return fmt.Errorf("no upstream lineage: the writes landed but nothing resolved")
}
for _, d := range lin.GetNodeDependencies() {
    fmt.Printf("%s -> %s\n",
        nodeName(lin.GetNodes()[d.GetSourceNodeIdx()]),
        nodeName(lin.GetNodes()[d.GetTargetNodeIdx()]),
    )
}

resp = api.lineage.GetLineage(
    lineage_service_pb2.GetLineageRequest(
        lineage_direction=lineage_direction_pb2.LINEAGE_DIRECTION_UPSTREAM,
        start_point=lineage_service_pb2.GetLineageStartPoint(
            entities=lineage_service_pb2.EntitiesStartPoint(
                entities=[custom_id("metabase::dashboard::9")]
            )
        ),
        max_depth=10,
    )
)
lin = resp.lineage
if not lin.node_dependencies:
    raise RuntimeError("no upstream lineage: the writes landed but nothing resolved")
for d in lin.node_dependencies:
    print(f"{node_name(lin.nodes[d.source_node_idx])} -> "
          f"{node_name(lin.nodes[d.target_node_idx])}")

A node carries every identifier that resolves to it, so nodeName picks the shape that says what the node is — prefer the custom variant when present. Check for the edges you expect, not merely that some edge came back. A declared column edge naming a column the upstream does not have still produces one hop while the rest of the chain is lost, so a non-empty result can still be a broken graph. Poll with a deadline, and exit non-zero when it expires rather than logging and returning. For column-level lineage, start from EntityColumnsStartPoint instead, naming the entity and the columns. The response then populates column_dependencies, and each node’s cll_details.cll_state says whether the parse succeeded. A state of RESOLUTION_FAILED on a node is usually a name your SqlDefinition should have bound and did not. This is not a first-version-only step. Keep it in the production integration.

Checklist before you ship

Ids are lowercase slugs from the source system’s stable id, in A-Z a-z 0-9 : _ . -
The title and any other human-readable text is in name, never in an id
Every type declares its traits, and its number was checked with ListTypes first
Every entity with columns declares its schema, before any lineage is written
Every name in a SqlDefinition that is not a warehouse object has a binding
Every upstream resolves, checked with BatchResolveIdentifiers (check_existence: true)
Check entities send package and kind, and leave both categories unset
One group holds everything, sent whole on every run, never from a partial read
feature_id values are stable strings, not generated per run
The program reads its own lineage back and fails loudly when it is missing

API surface

Every method, the scope it needs, and the limits

Agent workflow

A page to point a coding agent at, so it writes the integration for you

​Credentials

​1. Types

​2. Entities

​3. Schemas

​4. Lineage features

​SqlDefinition, with bindings

​ColumnLineage

​Code and Git

​5. Relationships

​6. Checks

​7. The entity group

​8. Executions

​9. Read it back

​Checklist before you ship

​Next

API surface

Agent workflow

Credentials

1. Types

2. Entities

3. Schemas

4. Lineage features

SqlDefinition, with bindings

ColumnLineage

Code and Git

5. Relationships

6. Checks

7. The entity group

8. Executions

9. Read it back

Checklist before you ship

Next