-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(strm-2706): merge with alpha branch
- Loading branch information
Showing
54 changed files
with
1,728 additions
and
640 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,7 @@ | ||
| .PHONY: clean-common-protos | ||
|
|
||
| SHELL := /bin/bash | ||
|
|
||
| common_protos := ${CURDIR}/.common-protos | ||
|
|
||
| grpc_version := 1.50.0 | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
app/src/main/kotlin/com/getstrm/pace/bigquery/BigQueryClient.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
189 changes: 92 additions & 97 deletions
189
app/src/main/kotlin/com/getstrm/pace/catalogs/Collibra.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,97 +1,92 @@ | ||
| //package com.getstrm.pace.catalogs | ||
| // | ||
| //import build.buf.gen.getstrm.api.data_policies.v1alpha.DataPolicy | ||
| //import com.apollographql.apollo3.ApolloClient | ||
| //import com.collibra.generated.ListPhysicalDataAssetsQuery | ||
| //import com.collibra.generated.ListSchemaIdsQuery | ||
| //import com.collibra.generated.ListTablesInSchemaQuery | ||
| //import com.collibra.generated.TableWithColumnsQuery | ||
| //import com.getstrm.pace.config.CatalogConfiguration | ||
| //import com.getstrm.pace.domain.DataCatalog | ||
| //import java.util.* | ||
| // | ||
| //class CollibraCatalog(config: CatalogConfiguration) : DataCatalog() { | ||
| // private val client = config.apolloClient() | ||
| // override fun close() { | ||
| // client.close() | ||
| // } | ||
| // | ||
| // override suspend fun listDatabases(): List<DataCatalog.Database> = listPhysicalAssets(AssetTypes.DATABASE).map { | ||
| // Database(this, it.id, it.getDataSourceType()) | ||
| // } | ||
| // | ||
| // class Database(private val catalog: CollibraCatalog, id: String, dbType: String) : | ||
| // DataCatalog.Database(id, dbType) { | ||
| // constructor(catalog: CollibraCatalog, id: Any, dbType: String) : this(catalog, id.toString(), dbType) | ||
| // | ||
| // override suspend fun getSchemas(): List<DataCatalog.Schema> { | ||
| // val assets = catalog.client.query(ListSchemaIdsQuery(id)).execute().data!!.assets!!.filterNotNull() | ||
| // .flatMap { schema -> | ||
| // schema.schemas | ||
| // } | ||
| // return assets.map { | ||
| // Schema(catalog, this, it.target.id.toString(), it.target.fullName) | ||
| // } | ||
| // } | ||
| // } | ||
| // | ||
| // class Schema(private val catalog: CollibraCatalog, database: DataCatalog.Database, id: String, name: String) : | ||
| // DataCatalog.Schema(database, id, name) { | ||
| // override suspend fun getTables(): List<DataCatalog.Table> = | ||
| // catalog.client.query(ListTablesInSchemaQuery(id)).execute().data!!.assets!!.filterNotNull() | ||
| // .flatMap { table -> | ||
| // table.tables.map { Table(catalog, this, it.target.id.toString(), it.target.fullName) } | ||
| // } | ||
| // } | ||
| // | ||
| // class Table(private val catalog: CollibraCatalog, schema: DataCatalog.Schema, id: String, name: String) : | ||
| // DataCatalog.Table(schema, id, name) { | ||
| // override suspend fun getDataPolicy(): DataPolicy? { | ||
| // val response = catalog.client.query(TableWithColumnsQuery(id = id)).execute() | ||
| // return response.data?.tables?.firstOrNull()?.let { table -> | ||
| // val systemName = | ||
| // table.schema.firstOrNull()?.schemaDetails?.database?.firstOrNull()?.databaseDetails?.domain?.name | ||
| // | ||
| // val builder = DataPolicy.newBuilder() | ||
| // builder.infoBuilder.title = table.displayName | ||
| // builder.infoBuilder.description = systemName | ||
| // builder.sourceBuilder.addAllAttributes(table.columns.map { it.toAttribute() }) | ||
| // builder.build() | ||
| // } | ||
| // } | ||
| // | ||
| // private fun TableWithColumnsQuery.Column.toAttribute(): DataPolicy.Attribute { | ||
| // return with(DataPolicy.Attribute.newBuilder()) { | ||
| // addPathComponents(columnDetails.displayName) | ||
| // val sourceType = columnDetails.dataType.firstOrNull()?.value ?: "unknown" | ||
| // // source type mapping | ||
| // type = sourceType | ||
| // build() | ||
| // } | ||
| // } | ||
| // } | ||
| // | ||
| // class Configuration( | ||
| // private val serverUrl: String = "https://test-drive.collibra.com/graphql/knowledgeGraph/v1", | ||
| // private val username: String = "test-drive-user-9b8o5m7l", | ||
| // private val password: String = "Egwrazg\$8q3j6i0b", | ||
| // ) : DataCatalog.Configuration() { | ||
| // fun apolloClient(): ApolloClient { | ||
| // val basicAuth = Base64.getEncoder().encodeToString("$username:$password".toByteArray()) | ||
| // | ||
| // return ApolloClient.Builder().serverUrl(serverUrl).addHttpHeader("Authorization", "Basic $basicAuth") | ||
| // .build() | ||
| // } | ||
| // } | ||
| // | ||
| // private suspend fun listPhysicalAssets(type: AssetTypes): List<ListPhysicalDataAssetsQuery.Asset> = | ||
| // client.query(ListPhysicalDataAssetsQuery(assetType = type.assetName)).execute().data!!.assets?.filterNotNull() | ||
| // ?: emptyList() | ||
| // | ||
| // private fun ListPhysicalDataAssetsQuery.Asset.getDataSourceType(): String = | ||
| // stringAttributes.find { it.type.publicId == "DataSourceType" }?.stringValue ?: "unknown" | ||
| //} | ||
| // | ||
| //enum class AssetTypes(val assetName: String) { | ||
| // DATABASE("Database"), SCHEMA("Schema"), TABLE("Table"), COLUMN("Column"), | ||
| //} | ||
| package com.getstrm.pace.catalogs | ||
| import build.buf.gen.getstrm.api.data_policies.v1alpha.DataPolicy | ||
| import com.apollographql.apollo3.ApolloClient | ||
| import com.collibra.generated.ListPhysicalDataAssetsQuery | ||
| import com.collibra.generated.ListSchemaIdsQuery | ||
| import com.collibra.generated.ListTablesInSchemaQuery | ||
| import com.collibra.generated.TableWithColumnsQuery | ||
| import com.getstrm.pace.config.CatalogConfiguration | ||
| import com.getstrm.pace.domain.DataCatalog | ||
| import normalizeType | ||
| import java.util.* | ||
|
|
||
| class CollibraCatalog(config: CatalogConfiguration) : DataCatalog(config) { | ||
|
|
||
| val client = apolloClient() | ||
| override fun close() { | ||
| client.close() | ||
| } | ||
|
|
||
| override suspend fun listDatabases(): List<DataCatalog.Database> = | ||
| listPhysicalAssets(AssetTypes.DATABASE).map { | ||
| Database(this, it.id, it.getDataSourceType()) | ||
| } | ||
|
|
||
| class Database(override val catalog: CollibraCatalog, id: String, dbType: String) : | ||
| DataCatalog.Database(catalog, id, dbType) { | ||
| constructor(catalog: CollibraCatalog, id: Any, dbType: String) : this(catalog, id.toString(), dbType) | ||
|
|
||
| override suspend fun getSchemas(): List<DataCatalog.Schema> { | ||
| val assets = catalog.client.query(ListSchemaIdsQuery(id)).execute().data!!.assets!!.filterNotNull().flatMap { schema -> | ||
| schema.schemas | ||
| } | ||
| return assets.map { | ||
| Schema(catalog, this, it.target.id.toString(), it.target.fullName) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class Schema(private val catalog: CollibraCatalog, database: DataCatalog.Database, id: String, name: String) : | ||
| DataCatalog.Schema(database, id, name) { | ||
| override suspend fun getTables(): List<DataCatalog.Table> = | ||
| catalog.client.query(ListTablesInSchemaQuery(id)).execute().data!!.assets!!.filterNotNull().flatMap { table -> | ||
| table.tables.map { Table(catalog, this, it.target.id.toString(), it.target.fullName) } | ||
| } | ||
| } | ||
|
|
||
| class Table(private val catalog: CollibraCatalog, schema: DataCatalog.Schema, id: String, name: String) : | ||
| DataCatalog.Table(schema, id, name) { | ||
| override suspend fun getDataPolicy(): DataPolicy? { | ||
| val response = catalog.client.query(TableWithColumnsQuery(id = id)).execute() | ||
| return response.data?.tables?.firstOrNull()?.let { table -> | ||
| val systemName = table.schema.firstOrNull()?.schemaDetails?.database?.firstOrNull()?.databaseDetails?.domain?.name | ||
|
|
||
| val builder = DataPolicy.newBuilder() | ||
| builder.infoBuilder.title = table.displayName | ||
| builder.infoBuilder.description = systemName | ||
| builder.sourceBuilder.addAllAttributes(table.columns.map { it.toAttribute() }) | ||
| builder.build() | ||
| } | ||
| } | ||
|
|
||
| private fun TableWithColumnsQuery.Column.toAttribute(): DataPolicy.Attribute = | ||
| with(DataPolicy.Attribute.newBuilder()) { | ||
| addPathComponents(columnDetails.displayName) | ||
| val sourceType = columnDetails.dataType.firstOrNull()?.value ?: "unknown" | ||
| // source type mapping | ||
| type = sourceType | ||
| addAllTags(columnDetails.tags.map { it.name }) | ||
| build().normalizeType() | ||
| } | ||
| } | ||
|
|
||
| private fun apolloClient(): ApolloClient { | ||
| val basicAuth = Base64.getEncoder().encodeToString("${config.userName}:${config.password}".toByteArray()) | ||
| return ApolloClient.Builder() | ||
| .serverUrl(config.serverUrl) | ||
| .addHttpHeader("Authorization", "Basic $basicAuth") | ||
| .build() | ||
| } | ||
| private suspend fun listPhysicalAssets(type: AssetTypes): List<ListPhysicalDataAssetsQuery.Asset> = | ||
| client.query(ListPhysicalDataAssetsQuery(assetType = type.assetName)).execute().data!!.assets?.filterNotNull() ?: emptyList() | ||
|
|
||
| private fun ListPhysicalDataAssetsQuery.Asset.getDataSourceType(): String = | ||
| stringAttributes.find { it.type.publicId == "DataSourceType" }?.stringValue ?: "unknown" | ||
| } | ||
|
|
||
| enum class AssetTypes(val assetName: String) { | ||
| DATABASE("Database"), | ||
| SCHEMA("Schema"), | ||
| TABLE("Table"), | ||
| COLUMN("Column"), | ||
| } |
Oops, something went wrong.