feat(ingest/sigma): Sigma connector integration

datahub-web-react/src/app/ingest/source/builder/constants.ts

@@ -32,6 +32,7 @@ import dynamodbLogo from '../../../../images/dynamodblogo.png';
 import fivetranLogo from '../../../../images/fivetranlogo.png';
 import csvLogo from '../../../../images/csv-logo.png';
 import qlikLogo from '../../../../images/qliklogo.png';
+import sigmaLogo from '../../../../images/sigmalogo.png';
 
 export const ATHENA = 'athena';
 export const ATHENA_URN = `urn:li:dataPlatform:${ATHENA}`;
@@ -116,6 +117,8 @@ export const CSV = 'csv-enricher';
 export const CSV_URN = `urn:li:dataPlatform:${CSV}`;
 export const QLIK_SENSE = 'qlik-sense';
 export const QLIK_SENSE_URN = `urn:li:dataPlatform:${QLIK_SENSE}`;
+export const SIGMA = 'sigma';
+export const SIGMA_URN = `urn:li:dataPlatform:${SIGMA}`;
 
 export const PLATFORM_URN_TO_LOGO = {
     [ATHENA_URN]: athenaLogo,
@@ -153,6 +156,7 @@ export const PLATFORM_URN_TO_LOGO = {
     [FIVETRAN_URN]: fivetranLogo,
     [CSV_URN]: csvLogo,
     [QLIK_SENSE_URN]: qlikLogo,
+    [SIGMA_URN]: sigmaLogo,
 };
 
 export const SOURCE_TO_PLATFORM_URN = {

datahub-web-react/src/app/ingest/source/builder/sources.json

@@ -244,4 +244,11 @@
         "docsUrl": "https://datahubproject.io/docs/generated/ingestion/sources/qlik-sense/",
         "recipe": "source:\n    type: qlik-sense\n    config:\n        # Coordinates\n        tenant_hostname: https://xyz12xz.us.qlikcloud.com\n        # Coordinates\n        api_key: QLIK_API_KEY\n\n        # Optional - filter for certain space names instead of ingesting everything.\n        # space_pattern:\n\n        #   allow:\n        #     - space_name\n        ingest_owner: true"
     }
+    {
+        "urn": "urn:li:dataPlatform:sigma",
+        "name": "sigma",
+        "displayName": "Sigma",
+        "docsUrl": "https://datahubproject.io/docs/generated/ingestion/sources/sigma/",
+        "recipe": "source:\n    type: sigma\n    config:\n        # Coordinates\n        api_url: https://aws-api.sigmacomputing.com/v2\n        # Coordinates\n        client_id: CLIENT_ID\n        client_secret: CLIENT_SECRET\n\n        # Optional - filter for certain workspace names instead of ingesting everything.\n        # workspace_pattern:\n\n        #   allow:\n        #     - workspace_name\n        ingest_owner: true"
+    }
 ]

metadata-ingestion/docs/sources/sigma/sigma_pre.md

@@ -0,0 +1,23 @@
+## Integration Details
+
+This source extracts the following:
+
+- Workspaces and workbooks within that workspaces as Container.
+- Sigma Datasets as Datahub Datasets.
+- Pages as Datahub dashboards and elements present inside pages as charts.
+
+## Configuration Notes
+
+1. Refer [doc](https://help.sigmacomputing.com/docs/generate-api-client-credentials) to generate an API client credentials.
+2. Provide the generated Client ID and Secret in Recipe.
+
+## Concept mapping 
+
+| Sigma                  | Datahub												         | Notes                            |
+|------------------------|---------------------------------------------------------------|----------------------------------|
+| `Workspace`            | [Container](../../metamodel/entities/container.md)     	     | SubType `"Sigma Workspace"`      |
+| `Workbook`             | [Container](../../metamodel/entities/container.md)            | SubType `"Sigma Workbook"`       |
+| `Page`                 | [Dashboard](../../metamodel/entities/dashboard.md)            |                                  |
+| `Element`              | [Chart](../../metamodel/entities/chart.md)                    |                                  |
+| `Dataset`              | [Dataset](../../metamodel/entities/dataset.md)                | SubType `"Sigma Dataset"`        |
+| `User`                 | [User (a.k.a CorpUser)](../../metamodel/entities/corpuser.md) | Optionally Extracted             |

metadata-ingestion/docs/sources/sigma/sigma_recipe.yml

@@ -0,0 +1,18 @@
+source:
+  type: sigma
+  config:
+    # Coordinates
+    api_url: "https://aws-api.sigmacomputing.com/v2"
+    # Credentials
+    client_id: "CLIENTID"
+    client_secret: "CLIENT_SECRET"
+
+    # Optional - filter for certain workspace names instead of ingesting everything.
+    # workspace_pattern:
+    #   allow:
+    #     - workspace_name
+
+    ingest_owner: true
+
+sink:
+  # sink configs

metadata-ingestion/setup.py

@@ -411,6 +411,7 @@
     "databricks": databricks | sql_common | sqllineage_lib,
     "fivetran": snowflake_common | bigquery_common,
     "qlik-sense": sqlglot_lib | {"requests", "websocket-client"},
+    "sigma": {"requests"},
 }
 
 # This is mainly used to exclude plugins from the Docker image.
@@ -546,6 +547,7 @@
             "fivetran",
             "kafka-connect",
             "qlik-sense",
+            "sigma",
         ]
         if plugin
         for dependency in plugins[plugin]
@@ -652,6 +654,7 @@
         "sql-queries = datahub.ingestion.source.sql_queries:SqlQueriesSource",
         "fivetran = datahub.ingestion.source.fivetran.fivetran:FivetranSource",
         "qlik-sense = datahub.ingestion.source.qlik_sense.qlik_sense:QlikSenseSource",
+        "sigma = datahub.ingestion.source.sigma.sigma:SigmaSource",
     ],
     "datahub.ingestion.transformer.plugins": [
         "pattern_cleanup_ownership = datahub.ingestion.transformer.pattern_cleanup_ownership:PatternCleanUpOwnership",

metadata-ingestion/src/datahub/ingestion/source/common/subtypes.py

@@ -17,6 +17,7 @@ class DatasetSubTypes(str, Enum):
     POWERBI_DATASET_TABLE = "PowerBI Dataset Table"
     QLIK_DATASET = "Qlik Dataset"
     BIGQUERY_TABLE_SNAPSHOT = "Bigquery Table Snapshot"
+    SIGMA_DATASET = "Sigma Dataset"
 
     # TODO: Create separate entity...
     NOTEBOOK = "Notebook"
@@ -43,6 +44,8 @@ class BIContainerSubTypes(str, Enum):
     POWERBI_DATASET = "PowerBI Dataset"
     QLIK_SPACE = "Qlik Space"
     QLIK_APP = "Qlik App"
+    SIGMA_WORKSPACE = "Sigma Workspace"
+    SIGMA_WORKBOOK = "Sigma Workbook"
 
 
 class BIAssetSubTypes(str, Enum):

metadata-ingestion/src/datahub/ingestion/source/sigma/config.py

@@ -0,0 +1,90 @@
+import logging
+from dataclasses import dataclass
+from typing import Dict, Optional
+
+import pydantic
+
+from datahub.configuration.common import AllowDenyPattern
+from datahub.configuration.source_common import (
+    EnvConfigMixin,
+    PlatformInstanceConfigMixin,
+)
+from datahub.ingestion.source.state.stale_entity_removal_handler import (
+    StaleEntityRemovalSourceReport,
+)
+from datahub.ingestion.source.state.stateful_ingestion_base import (
+    StatefulIngestionConfigBase,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class Constant:
+    """
+    keys used in sigma plugin
+    """
+
+    # Rest API response key constants
+    ENTRIES = "entries"
+    FIRSTNAME = "firstName"
+    LASTNAME = "lastName"
+    EDGES = "edges"
+    DEPENDENCIES = "dependencies"
+    SOURCE = "source"
+    WORKSPACEID = "workspaceId"
+    PATH = "path"
+    NAME = "name"
+    URL = "url"
+    ELEMENTID = "elementId"
+    ID = "id"
+    PARENTID = "parentId"
+    TYPE = "type"
+    DATASET = "dataset"
+    WORKBOOK = "workbook"
+    BADGE = "badge"
+    NEXTPAGE = "nextPage"
+
+    # Source Config constants
+    DEFAULT_API_URL = "https://aws-api.sigmacomputing.com/v2"
+
+
+@dataclass
+class SigmaSourceReport(StaleEntityRemovalSourceReport):
+    number_of_workspaces: int = 0
+
+    def report_number_of_workspaces(self, number_of_workspaces: int) -> None:
+        self.number_of_workspaces = number_of_workspaces
+
+
+class PlatformDetail(PlatformInstanceConfigMixin, EnvConfigMixin):
+    data_source_platform: str = pydantic.Field(
+        description="A chart's data sources platform name.",
+    )
+
+
+class SigmaSourceConfig(
+    StatefulIngestionConfigBase, PlatformInstanceConfigMixin, EnvConfigMixin
+):
+    api_url: str = pydantic.Field(
+        default=Constant.DEFAULT_API_URL, description="Sigma API hosted URL."
+    )
+    client_id: str = pydantic.Field(description="Sigma Client ID")
+    client_secret: str = pydantic.Field(description="Sigma Client Secret")
+    # Sigma workspace identifier
+    workspace_pattern: AllowDenyPattern = pydantic.Field(
+        default=AllowDenyPattern.allow_all(),
+        description="Regex patterns to filter Sigma workspaces in ingestion."
+        "Mention 'User Folder' if entities of 'My documents' need to ingest.",
+    )
+    ingest_owner: Optional[bool] = pydantic.Field(
+        default=True,
+        description="Ingest Owner from source. This will override Owner info entered from UI",
+    )
+    chart_sources_platform_mapping: Dict[str, PlatformDetail] = pydantic.Field(
+        default={},
+        description="A mapping of the sigma workspace/workbook/chart folder path to all chart's data sources platform details present inside that folder path. "
+        "For example: Provide key as 'workspace_name/[folder_name/]workbook_name/chart_name' if just for one specific chart, "
+        "key as 'workspace_name/[folder_name/]workbook_name' if for all charts within a specific workbook, "
+        "key as 'workspace_name' if for all workbook's charts within a specific workspace, and "
+        "key as '*' for all ingested charts",
+    )

metadata-ingestion/src/datahub/ingestion/source/sigma/data_classes.py

@@ -0,0 +1,73 @@
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel, root_validator
+
+from datahub.emitter.mcp_builder import ContainerKey
+
+
+class WorkspaceKey(ContainerKey):
+    workspaceId: str
+
+
+class WorkbookKey(ContainerKey):
+    workbookId: str
+
+
+class Workspace(BaseModel):
+    workspaceId: str
+    name: str
+    createdBy: str
+    createdAt: datetime
+    updatedAt: datetime
+
+
+class SigmaDataset(BaseModel):
+    datasetId: str
+    workspaceId: str
+    name: str
+    description: str
+    createdBy: str
+    createdAt: datetime
+    updatedAt: datetime
+    url: str
+    path: str
+    badge: Optional[str] = None
+
+    @root_validator(pre=True)
+    def update_values(cls, values: Dict) -> Dict:
+        # As element lineage api provide this id as source dataset id
+        values["datasetId"] = values["url"].split("/")[-1]
+        return values
+
+
+class Element(BaseModel):
+    elementId: str
+    type: str
+    name: str
+    url: str
+    vizualizationType: Optional[str] = None
+    query: Optional[str] = None
+    columns: List[str] = []
+    upstream_sources: Dict[str, str] = {}
+
+
+class Page(BaseModel):
+    pageId: str
+    name: str
+    elements: List[Element] = []
+
+
+class Workbook(BaseModel):
+    workbookId: str
+    workspaceId: str
+    name: str
+    createdBy: str
+    updatedBy: str
+    createdAt: datetime
+    updatedAt: datetime
+    url: str
+    path: str
+    latestVersion: int
+    pages: List[Page] = []
+    badge: Optional[str] = None