Changes for no columns and added owner for table (#11086)

* changes for no columns and added owner for table

* added pydantic model for owners

ingestion/src/metadata/ingestion/source/database/domodatabase/metadata.py

@@ -14,7 +14,7 @@ Domo Database source to extract metadata
 """
 
 import traceback
-from typing import Iterable, Optional, Tuple
+from typing import Iterable, List, Optional, Tuple
 
 from metadata.clients.domo_client import DomoClient
 from metadata.generated.schema.api.data.createDatabase import CreateDatabaseRequest
@@ -36,11 +36,18 @@ from metadata.generated.schema.metadataIngestion.databaseServiceMetadataPipeline
 from metadata.generated.schema.metadataIngestion.workflow import (
     Source as WorkflowSource,
 )
+from metadata.generated.schema.type.entityReference import EntityReference
 from metadata.ingestion.api.source import InvalidSourceException
 from metadata.ingestion.models.ometa_classification import OMetaTagAndClassification
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.connections import get_connection
 from metadata.ingestion.source.database.database_service import DatabaseServiceSource
+from metadata.ingestion.source.database.domodatabase.models import (
+    OutputDataset,
+    Owner,
+    SchemaColumn,
+    User,
+)
 from metadata.utils import fqn
 from metadata.utils.constants import DEFAULT_DATABASE
 from metadata.utils.filters import filter_by_table
@@ -135,20 +142,36 @@ class DomodatabaseSource(DatabaseServiceSource):
             logger.warning(error)
             self.status.failed(schema_name, error, traceback.format_exc())
 
+    def get_owners(self, owner: Owner) -> Optional[EntityReference]:
+        try:
+            owner_details = User(**self.domo_client.users_get(owner.id))
+            if owner_details.email:
+                user = self.metadata.get_user_by_email(owner_details.email)
+                if user:
+                    return EntityReference(id=user.id.__root__, type="user")
+        except Exception as exc:
+            logger.warning(f"Error while getting details of user {owner.name} - {exc}")
+        return None
+
     def yield_table(
         self, table_name_and_type: Tuple[str, str]
     ) -> Iterable[Optional[CreateTableRequest]]:
         table_id, table_type = table_name_and_type
         try:
             table_constraints = None
-            table_object = self.domo_client.datasets.get(table_id)
-            columns = self.get_columns(table_object=table_object["schema"]["columns"])
+            table_object = OutputDataset(**self.domo_client.datasets.get(table_id))
+            columns = (
+                self.get_columns(table_object.schemas.columns)
+                if table_object.columns
+                else []
+            )
             table_request = CreateTableRequest(
-                name=table_object["name"],
-                displayName=table_object["name"],
+                name=table_object.name,
+                displayName=table_object.name,
                 tableType=table_type,
-                description=table_object.get("description"),
+                description=table_object.description,
                 columns=columns,
+                owner=self.get_owners(owner=table_object.owner),
                 tableConstraints=table_constraints,
                 databaseSchema=self.context.database_schema.fullyQualifiedName,
             )
@@ -160,15 +183,15 @@ class DomodatabaseSource(DatabaseServiceSource):
             logger.warning(error)
             self.status.failed(table_id, error, traceback.format_exc())
 
-    def get_columns(self, table_object):
+    def get_columns(self, table_object: List[SchemaColumn]):
         row_order = 1
         columns = []
         for column in table_object:
             columns.append(
                 Column(
-                    name=column["name"],
-                    description=column.get("description", ""),
-                    dataType=column["type"],
+                    name=column.name,
+                    description=column.description,
+                    dataType=column.type,
                     ordinalPosition=row_order,
                 )
             )

ingestion/src/metadata/ingestion/source/database/domodatabase/models.py

@@ -0,0 +1,54 @@
+#  Copyright 2021 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""
+Domo Database Source Model module
+"""
+
+from typing import List, Optional
+
+from pydantic import BaseModel, Extra, Field
+
+
+class DomoDatabaseBaseModel(BaseModel):
+    class Config:
+        extra = Extra.allow
+
+    id: str
+    name: str
+
+
+class User(DomoDatabaseBaseModel):
+    email: str
+    role: str
+
+
+class SchemaColumn(BaseModel):
+    type: str
+    name: str
+    description: Optional[str]
+
+
+class Schema(BaseModel):
+    columns: List[SchemaColumn]
+
+
+class Owner(DomoDatabaseBaseModel):
+    id: int
+    name: str
+
+
+class OutputDataset(DomoDatabaseBaseModel):
+    rows: int
+    columns: int
+    schemas: Optional[Schema] = Field(alias="schema")
+    owner: Owner
+    description: Optional[str]