yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-11-08 15:30:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(ingest): use typing-aware deprecation type (#12982)

Browse Source
This commit is contained in:
Harshal Sheth 2025-03-26 11:32:46 -07:00 committed by GitHub
parent 98dc67fa4c
commit 1d22c58e13
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 9 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
metadata-ingestion/setup.py
View File

@ -15,7 +15,7 @@ _self_pin = (
base_requirements = { base_requirements = {
# Our min version of typing_extensions is somewhat constrained by Airflow. # Our min version of typing_extensions is somewhat constrained by Airflow.
"typing_extensions>=4.2.0", "typing_extensions>=4.5.0",
# Actual dependencies. # Actual dependencies.
"typing-inspect", "typing-inspect",
# pydantic 1.8.2 is incompatible with mypy 0.910. # pydantic 1.8.2 is incompatible with mypy 0.910.

4
metadata-ingestion/src/datahub/emitter/rest_emitter.py
View File

@ -24,9 +24,9 @@ from typing import (
import pydantic import pydantic
import requests import requests
from deprecated import deprecated
from requests.adapters import HTTPAdapter, Retry from requests.adapters import HTTPAdapter, Retry
from requests.exceptions import HTTPError, RequestException from requests.exceptions import HTTPError, RequestException
from typing_extensions import deprecated
from datahub._version import nice_version_name from datahub._version import nice_version_name
from datahub.cli import config_utils from datahub.cli import config_utils
@ -597,7 +597,7 @@ class DataHubRestEmitter(Closeable, Emitter):
return len(mcp_obj_chunks) return len(mcp_obj_chunks)
@deprecated @deprecated("Use emit with a datasetUsageStatistics aspect instead")
def emit_usage(self, usageStats: UsageAggregation) -> None: def emit_usage(self, usageStats: UsageAggregation) -> None:
url = f"{self._gms_server}/usageStats?action=batchIngest" url = f"{self._gms_server}/usageStats?action=batchIngest"

17
metadata-ingestion/src/datahub/ingestion/graph/client.py
View File

@ -23,9 +23,9 @@ from typing import (
) )
from avro.schema import RecordSchema from avro.schema import RecordSchema
from deprecated import deprecated
from pydantic import BaseModel from pydantic import BaseModel
from requests.models import HTTPError from requests.models import HTTPError
from typing_extensions import deprecated
from datahub.cli import config_utils from datahub.cli import config_utils
from datahub.configuration.common import ConfigModel, GraphError, OperationalError from datahub.configuration.common import ConfigModel, GraphError, OperationalError
@ -330,7 +330,7 @@ class DataHubGraph(DatahubRestEmitter, EntityVersioningAPI):
f"Failed to find {aspect_type_name} in response {response_json}" f"Failed to find {aspect_type_name} in response {response_json}"
) )
@deprecated(reason="Use get_aspect instead which makes aspect string name optional") @deprecated("Use get_aspect instead which makes aspect string name optional")
def get_aspect_v2( def get_aspect_v2(
self, self,
entity_urn: str, entity_urn: str,
@ -355,7 +355,7 @@ class DataHubGraph(DatahubRestEmitter, EntityVersioningAPI):
def get_schema_metadata(self, entity_urn: str) -> Optional[SchemaMetadataClass]: def get_schema_metadata(self, entity_urn: str) -> Optional[SchemaMetadataClass]:
return self.get_aspect(entity_urn=entity_urn, aspect_type=SchemaMetadataClass) return self.get_aspect(entity_urn=entity_urn, aspect_type=SchemaMetadataClass)
@deprecated(reason="Use get_aspect directly.") @deprecated("Use get_aspect directly.")
def get_domain_properties(self, entity_urn: str) -> Optional[DomainPropertiesClass]: def get_domain_properties(self, entity_urn: str) -> Optional[DomainPropertiesClass]:
return self.get_aspect(entity_urn=entity_urn, aspect_type=DomainPropertiesClass) return self.get_aspect(entity_urn=entity_urn, aspect_type=DomainPropertiesClass)
@ -376,7 +376,7 @@ class DataHubGraph(DatahubRestEmitter, EntityVersioningAPI):
def get_domain(self, entity_urn: str) -> Optional[DomainsClass]: def get_domain(self, entity_urn: str) -> Optional[DomainsClass]:
return self.get_aspect(entity_urn=entity_urn, aspect_type=DomainsClass) return self.get_aspect(entity_urn=entity_urn, aspect_type=DomainsClass)
@deprecated(reason="Use get_aspect directly.") @deprecated("Use get_aspect directly.")
def get_browse_path(self, entity_urn: str) -> Optional[BrowsePathsClass]: def get_browse_path(self, entity_urn: str) -> Optional[BrowsePathsClass]:
return self.get_aspect(entity_urn=entity_urn, aspect_type=BrowsePathsClass) return self.get_aspect(entity_urn=entity_urn, aspect_type=BrowsePathsClass)
@ -505,7 +505,7 @@ class DataHubGraph(DatahubRestEmitter, EntityVersioningAPI):
return response.json() return response.json()
@deprecated( @deprecated(
reason="Use get_aspect for a single aspect or get_entity_semityped for a full entity." "Use get_aspect for a single aspect or get_entity_semityped for a full entity."
) )
def get_aspects_for_entity( def get_aspects_for_entity(
self, self,
@ -635,9 +635,6 @@ class DataHubGraph(DatahubRestEmitter, EntityVersioningAPI):
def _aspect_count_endpoint(self): def _aspect_count_endpoint(self):
return f"{self.config.server}/aspects?action=getCount" return f"{self.config.server}/aspects?action=getCount"
# def _session(self) -> Session:
# return super()._session
def get_domain_urn_by_name(self, domain_name: str) -> Optional[str]: def get_domain_urn_by_name(self, domain_name: str) -> Optional[str]:
"""Retrieve a domain urn based on its name. Returns None if there is no match found""" """Retrieve a domain urn based on its name. Returns None if there is no match found"""
@ -749,9 +746,7 @@ class DataHubGraph(DatahubRestEmitter, EntityVersioningAPI):
assert res["upsertConnection"]["urn"] == urn assert res["upsertConnection"]["urn"] == urn
@deprecated( @deprecated('Use get_urns_by_filter(entity_types=["container"], ...) instead')
reason='Use get_urns_by_filter(entity_types=["container"], ...) instead'
)
def get_container_urns_by_filter( def get_container_urns_by_filter(
self, self,
env: Optional[str] = None, env: Optional[str] = None,