fmd-1081 - Add publication Dataset

Signed-off-by: Helder Ribeiro <[email protected]>

home/service/details.py

@@ -229,5 +229,32 @@ def _get_context(self):
 
         return context
 
+
 class PublicationDatasetDetailsService(GenericService):
-    pass
+    def __init__(self, urn: str):
+        self.urn = urn
+        self.client = self._get_catalogue_client()
+
+        self.publication_dataset_metadata = self.client.get_publication_dataset_details(self.urn)
+
+        if not self.publication_dataset_metadata:
+            raise ObjectDoesNotExist(urn)
+
+        relationships = self.publication_dataset_metadata.relationships or {}
+        self.parent_entity = _parse_parent(relationships)
+        self.context = self._get_context()
+        self.template = "details_publication_dataset.html"
+
+    def _get_context(self):
+
+        return {
+            "entity": self.publication_dataset_metadata,
+            "entity_type": "Table",
+            "parent_entity": self.parent_entity,
+            "parent_type": ResultType.DATABASE.name.lower(),
+            "h1_value": self.publication_dataset_metadata.name,
+            # noqa: E501
+            "is_access_requirements_a_url": is_access_requirements_a_url(
+                self.publication_dataset_metadata.custom_properties.access_information.dc_access_requirements
+            ),
+        }

lib/datahub-client/data_platform_catalogue/client/datahub_client.py

@@ -65,7 +65,7 @@
     Governance,
     RelationshipType,
     Table,
-    PublicationCollection,
+    PublicationCollection, PublicationDataset,
 )
 from data_platform_catalogue.search_types import (
     DomainOption,
@@ -450,6 +450,52 @@ def get_publication_collection_details(self, urn: str) -> PublicationCollection:
             )
         raise EntityDoesNotExist(f"Database with urn: {urn} does not exist")
 
+    def get_publication_dataset_details(self, urn: str) -> PublicationDataset:
+        if self.check_entity_exists_by_urn(urn):
+            response = self.graph.execute_graphql(self.dataset_query, {"urn": urn})[
+                "dataset"
+            ]
+            platform_name = response["platform"]["name"]
+            properties, custom_properties = parse_properties(response)
+            domain = parse_domain(response)
+            owner = parse_data_owner(response)
+            stewards = parse_stewards(response)
+            custodians = parse_custodians(response)
+            tags = parse_tags(response)
+            glossary_terms = parse_glossary_terms(response)
+            created, modified = parse_created_and_modified(properties)
+            modified = parse_last_modified(response)
+            name, display_name, qualified_name = parse_names(response, properties)
+
+            parent_relations = parse_relations(
+                RelationshipType.PARENT,
+                [response.get("parent_container_relations", {})],
+            )
+            parent_relations_to_display = self.list_relations_to_display(
+                parent_relations
+            )
+
+            return PublicationDataset(
+                urn=urn,
+                external_url=properties.get("externalUrl", ""),
+                display_name=display_name,
+                name=name,
+                fully_qualified_name=qualified_name,
+                description=properties.get("description", ""),
+                relationships={**parent_relations_to_display},
+                domain=domain,
+                governance=Governance(
+                    data_owner=owner, data_custodians=custodians, data_stewards=stewards
+                ),
+                tags=tags,
+                glossary_terms=glossary_terms,
+                last_modified=modified,
+                created=created,
+                custom_properties=custom_properties,
+                platform=EntityRef(display_name=platform_name, urn=platform_name),
+            )
+        raise EntityDoesNotExist(f"Database with urn: {urn} does not exist")
+
     def get_dashboard_details(self, urn: str) -> Dashboard:
         if self.check_entity_exists_by_urn(urn):
             response = self.graph.execute_graphql(self.dashboard_query, {"urn": urn})[

lib/datahub-client/data_platform_catalogue/client/graphql/getDatasetDetails.graphql

@@ -93,6 +93,7 @@ query getDatasetDetails($urn: String!) {
       name
       qualifiedName
       description
+      externalUrl
       customProperties {
         key
         value

lib/datahub-client/data_platform_catalogue/entities.py

@@ -500,6 +500,19 @@ class PublicationCollection(Entity):
         examples=["https://data.justice.gov.uk/prisons/criminal-jsutice/publications"],
     )
 
+
+class PublicationDataset(Entity):
+    """For source system publication collections"""
+    urn: str | None = Field(
+        description="Unique identifier for the entity. Relates to Datahub's urn",
+        examples=["urn:li:dataset:(urn:li:dataPlatform:gov.uk,statistics2011,DEV)"],
+    )
+    external_url: str = Field(
+        description="URL to view the collection",
+        examples=["https://data.justice.gov.uk/prisons/criminal-jsutice/publications"],
+    )
+
+
 class Table(Entity):
     """A table in a database or a tabular dataset. DataHub calls them datasets."""