atlan_integration
Find full example heremain.go
package main
import (
extensionsatlanintegrationsv1grpc "buf.build/gen/go/getsynq/api/grpc/go/synq/extensions/atlan/integrations/v1/integrationsv1grpc"
extensionsatlanintegrationsv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/extensions/atlan/integrations/v1"
"context"
"crypto/tls"
"encoding/json"
"fmt"
"golang.org/x/oauth2/clientcredentials"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/oauth"
"os"
)
func main() {
ctx := context.Background()
host := "developer.synq.io"
port := "443"
apiUrl := fmt.Sprintf("%s:%s", host, port)
clientID := os.Getenv("SYNQ_CLIENT_ID")
clientSecret := os.Getenv("SYNQ_CLIENT_SECRET")
tokenURL := fmt.Sprintf("https://%s/oauth2/token", host)
config := &clientcredentials.Config{
ClientID: clientID,
ClientSecret: clientSecret,
TokenURL: tokenURL,
}
oauthTokenSource := oauth.TokenSource{TokenSource: config.TokenSource(ctx)}
creds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: false})
opts := []grpc.DialOption{
grpc.WithTransportCredentials(creds),
grpc.WithPerRPCCredentials(oauthTokenSource),
grpc.WithAuthority(host),
}
conn, err := grpc.DialContext(ctx, apiUrl, opts...)
if err != nil {
panic(err)
}
defer conn.Close()
fmt.Printf("Connected to API...\n\n")
integrationsApi := extensionsatlanintegrationsv1grpc.NewAtlanIntegrationServiceClient(conn)
// Upsert tenant info and validate.
{
tenantUrl := os.Getenv("ATLAN_TENANT_URL")
tenantApiToken := os.Getenv("ATLAN_API_TOKEN")
resp, err := integrationsApi.Upsert(ctx, &extensionsatlanintegrationsv1.UpsertRequest{
AtlanTenantUrl: tenantUrl,
AtlanApiToken: tenantApiToken,
})
if err != nil {
panic(err)
}
if !resp.Integration.IsValid {
panic("integration not valid: connection to atlan failed")
}
}
// Fetch integration.
{
resp, err := integrationsApi.Get(ctx, &extensionsatlanintegrationsv1.GetRequest{})
if err != nil {
panic(err)
}
b, _ := json.Marshal(resp.Integration)
fmt.Printf("Existing integration -> %+v", string(b))
}
}
main.go
package main
import (
extensionsatlanproviderv1grpc "buf.build/gen/go/getsynq/api/grpc/go/synq/extensions/atlan/provider/v1/providerv1grpc"
extensionsatlanproviderv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/extensions/atlan/provider/v1"
"context"
"crypto/tls"
"fmt"
"golang.org/x/oauth2/clientcredentials"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/oauth"
"os"
)
func main() {
ctx := context.Background()
host := "developer.synq.io"
port := "443"
apiUrl := fmt.Sprintf("%s:%s", host, port)
clientID := os.Getenv("SYNQ_CLIENT_ID")
clientSecret := os.Getenv("SYNQ_CLIENT_SECRET")
tokenURL := fmt.Sprintf("https://%s/oauth2/token", host)
config := &clientcredentials.Config{
ClientID: clientID,
ClientSecret: clientSecret,
TokenURL: tokenURL,
}
oauthTokenSource := oauth.TokenSource{TokenSource: config.TokenSource(ctx)}
creds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: false})
opts := []grpc.DialOption{
grpc.WithTransportCredentials(creds),
grpc.WithPerRPCCredentials(oauthTokenSource),
grpc.WithAuthority(host),
}
conn, err := grpc.DialContext(ctx, apiUrl, opts...)
if err != nil {
panic(err)
}
defer conn.Close()
fmt.Printf("Connected to API...\n\n")
atlanProviderApi := extensionsatlanproviderv1grpc.NewAtlanProviderServiceClient(conn)
// Requires valid integration created in 1_setup_integration.
// Fetch visible connections, products and domains.
{
resp, err := atlanProviderApi.GetAtlanConnections(ctx, &extensionsatlanproviderv1.GetAtlanConnectionsRequest{})
if err != nil {
panic(err)
}
fmt.Printf("%d Visible Atlan Connections:\n", len(resp.Connections))
for _, connection := range resp.Connections {
fmt.Printf(" -> %+v\n", connection)
}
}
{
resp, err := atlanProviderApi.GetAtlanDataProducts(ctx, &extensionsatlanproviderv1.GetAtlanDataProductsRequest{})
if err != nil {
panic(err)
}
fmt.Printf("%d Visible Atlan DataProducts:\n", len(resp.DataProducts))
for _, dataProduct := range resp.DataProducts {
fmt.Printf(" -> %+v\n", dataProduct)
}
}
{
resp, err := atlanProviderApi.GetAtlanDomains(ctx, &extensionsatlanproviderv1.GetAtlanDomainsRequest{})
if err != nil {
panic(err)
}
fmt.Printf("%d Visible Atlan DataDomains:\n", len(resp.Domains))
for _, domain := range resp.Domains {
fmt.Printf(" -> %+v\n", domain)
}
}
}
main.go
package main
import (
extensionsatlanintegrationsv1grpc "buf.build/gen/go/getsynq/api/grpc/go/synq/extensions/atlan/integrations/v1/integrationsv1grpc"
extensionsatlanworkflowsv1grpc "buf.build/gen/go/getsynq/api/grpc/go/synq/extensions/atlan/workflows/v1/workflowsv1grpc"
extensionsatlanintegrationsv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/extensions/atlan/integrations/v1"
extensionsatlanworkflowsv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/extensions/atlan/workflows/v1"
platformsv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/platforms/v1"
"context"
"crypto/tls"
"encoding/json"
"fmt"
"golang.org/x/oauth2/clientcredentials"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/oauth"
"os"
)
func main() {
ctx := context.Background()
host := "developer.synq.io"
port := "443"
apiUrl := fmt.Sprintf("%s:%s", host, port)
clientID := os.Getenv("SYNQ_CLIENT_ID")
clientSecret := os.Getenv("SYNQ_CLIENT_SECRET")
tokenURL := fmt.Sprintf("https://%s/oauth2/token", host)
config := &clientcredentials.Config{
ClientID: clientID,
ClientSecret: clientSecret,
TokenURL: tokenURL,
}
oauthTokenSource := oauth.TokenSource{TokenSource: config.TokenSource(ctx)}
creds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: false})
opts := []grpc.DialOption{
grpc.WithTransportCredentials(creds),
grpc.WithPerRPCCredentials(oauthTokenSource),
grpc.WithAuthority(host),
}
conn, err := grpc.DialContext(ctx, apiUrl, opts...)
if err != nil {
panic(err)
}
defer conn.Close()
fmt.Printf("Connected to API...\n\n")
integrationsApi := extensionsatlanintegrationsv1grpc.NewAtlanIntegrationServiceClient(conn)
workflowsApi := extensionsatlanworkflowsv1grpc.NewAtlanWorkflowServiceClient(conn)
// Requires valid integration created in 1_setup_integration.
// Map Atlan connections to SYNQ integrations.
// Use 2_fetch_atlan_resources to find visible connections.
{
_, err := workflowsApi.SetConnectionMappings(ctx, &extensionsatlanworkflowsv1.SetConnectionMappingsRequest{
Mappings: []*extensionsatlanworkflowsv1.ConnectionMapping{
{
AtlanConnectionQualifiedName: "default/dbt/1",
SynqDataPlatformIdentifier: &platformsv1.DataPlatformIdentifier{
Id: &platformsv1.DataPlatformIdentifier_DbtCloud{
DbtCloud: &platformsv1.DbtCloudIdentifier{
ApiEndpoint: "cloud.getdbt.com",
AccountId: "1234",
ProjectId: "5678",
},
},
},
},
},
})
if err != nil {
panic(err)
}
}
// Activate integration once you are ready to synchronize.
// This runs a synchronization every 5 minutes.
{
_, err := integrationsApi.Activate(ctx, &extensionsatlanintegrationsv1.ActivateRequest{
Activate: true,
})
if err != nil {
panic(err)
}
}
// You can also manually synchronize with atlan.
// This creates the dataproducts and associated domains visible from Atlan into SYNQ
{
resp, err := workflowsApi.Synchronize(ctx, &extensionsatlanworkflowsv1.SynchronizeRequest{})
if err != nil {
panic(err)
}
b, _ := json.Marshal(resp.WorkflowRun)
fmt.Printf("Synchronization result -> %s\n\n", string(b))
if resp.HasErrors {
panic("synchronization has errors")
}
}
// Fetch mapped products and domains.
{
resp, err := workflowsApi.GetDomainMappings(ctx, &extensionsatlanworkflowsv1.GetDomainMappingsRequest{})
if err != nil {
panic(err)
}
b, _ := json.Marshal(resp.Mappings)
fmt.Printf("Mapped domains -> %s\n\n", string(b))
}
{
resp, err := workflowsApi.GetProductMappings(ctx, &extensionsatlanworkflowsv1.GetProductMappingsRequest{})
if err != nil {
panic(err)
}
b, _ := json.Marshal(resp.Mappings)
fmt.Printf("Mapped products -> %s\n\n", string(b))
}
// Fetch synchronization runs history.
{
resp, err := workflowsApi.FetchRuns(ctx, &extensionsatlanworkflowsv1.FetchRunsRequest{
From: 0,
Limit: 10,
})
if err != nil {
panic(err)
}
fmt.Printf("Found %d runs.\n", len(resp.WorkflowRuns))
for _, run := range resp.WorkflowRuns {
b, _ := json.Marshal(run)
fmt.Printf(" -> [%+v] (%+v) %s\n", run.StartedAt, run.Status, string(b))
}
}
}
main.go
package main
import (
extensionsatlanintegrationsv1grpc "buf.build/gen/go/getsynq/api/grpc/go/synq/extensions/atlan/integrations/v1/integrationsv1grpc"
extensionsatlanintegrationsv1 "buf.build/gen/go/getsynq/api/protocolbuffers/go/synq/extensions/atlan/integrations/v1"
"context"
"crypto/tls"
"fmt"
"golang.org/x/oauth2/clientcredentials"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/oauth"
"os"
)
func main() {
ctx := context.Background()
host := "developer.synq.io"
port := "443"
apiUrl := fmt.Sprintf("%s:%s", host, port)
clientID := os.Getenv("SYNQ_CLIENT_ID")
clientSecret := os.Getenv("SYNQ_CLIENT_SECRET")
tokenURL := fmt.Sprintf("https://%s/oauth2/token", host)
config := &clientcredentials.Config{
ClientID: clientID,
ClientSecret: clientSecret,
TokenURL: tokenURL,
}
oauthTokenSource := oauth.TokenSource{TokenSource: config.TokenSource(ctx)}
creds := credentials.NewTLS(&tls.Config{InsecureSkipVerify: false})
opts := []grpc.DialOption{
grpc.WithTransportCredentials(creds),
grpc.WithPerRPCCredentials(oauthTokenSource),
grpc.WithAuthority(host),
}
conn, err := grpc.DialContext(ctx, apiUrl, opts...)
if err != nil {
panic(err)
}
defer conn.Close()
fmt.Printf("Connected to API...\n\n")
integrationsApi := extensionsatlanintegrationsv1grpc.NewAtlanIntegrationServiceClient(conn)
// Deactivate integraton.
// This stops the scheduled synchronization.
{
_, err := integrationsApi.Activate(ctx, &extensionsatlanintegrationsv1.ActivateRequest{
Activate: false,
})
if err != nil {
panic(err)
}
fmt.Println("Deactivated Atlan integration.")
}
// Optionally delete integration.
{
_, err := integrationsApi.Remove(ctx, &extensionsatlanintegrationsv1.RemoveRequest{})
if err != nil {
panic(err)
}
fmt.Println("Removed Atlan integration.")
}
// Note that the domains and products created on SYNQ by the synchronization
// are *NOT* automatically removed. You can remove them from SYNQ if you wish to.
}