def get_workunits_internal()

in sync/datahub/glean_source.py [0:0]

• 53 lines of code
• 4 McCabe index (conditional complexity)

    def get_workunits_internal(self) -> Iterable[MetadataWorkUnit]:
        for glean_ping in get_glean_pings():
            glean_qualified_urn = builder.make_dataset_urn(
                platform=self.platform,
                name=glean_ping.qualified_name,
                env=self.config.env,
            )
            glean_ping_aspects = [
                InstitutionalMemoryClass(
                    elements=[
                        InstitutionalMemoryMetadataClass(
                            url=glean_ping.glean_dictionary_url,
                            description="Glean Dictionary Ping Documentation",
                            createStamp=get_current_timestamp(),
                        )
                    ],
                ),
                DatasetPropertiesClass(
                    name=glean_ping.name, description=glean_ping.description
                ),
                SubTypesClass(typeNames=["Ping"]),
                BrowsePathsClass(
                    paths=[f"/{self.config.env.lower()}/glean/{glean_ping.app_name}"]
                ),
            ]
            glean_ping_mcps = MetadataChangeProposalWrapper.construct_many(
                entityUrn=glean_qualified_urn, aspects=glean_ping_aspects
            )

            upstream_lineage = UpstreamLineageClass(
                upstreams=[
                    UpstreamClass(
                        dataset=glean_qualified_urn,
                        type="TRANSFORMED",
                    )
                ]
            )
            upstream_lineage_mcps = [
                MetadataChangeProposalWrapper(
                    entityType="dataset",
                    changeType=ChangeTypeClass.UPSERT,
                    entityUrn=builder.make_dataset_urn(
                        platform="bigquery",
                        name=qualified_table_name,
                        env=self.config.env,
                    ),
                    aspectName="upstreamLineage",
                    aspect=upstream_lineage,
                )
                for qualified_table_name in glean_ping.bigquery_fully_qualified_names
            ]

            for mcp in glean_ping_mcps + upstream_lineage_mcps:
                wu = mcp.as_workunit()
                yield wu