Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

docs(ingest): inherit capabilities from superclasses #9174

Merged
merged 3 commits into from Nov 6, 2023
Merged

docs(ingest): inherit capabilities from superclasses #9174

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e605e27
docs(ingest): inherit capabilities from superclasses
hsheth2 Nov 3, 2023
3381d5f
fix lint
hsheth2 Nov 3, 2023
f45e3cc
fix airflow tests
hsheth2 Nov 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 4 additions & 0 deletions metadata-ingestion-modules/airflow-plugin/setup.py
View file
Open in desktop
Expand Up @@ -101,6 +101,10 @@ def get_long_description():
f"acryl-datahub[testing-utils]{_self_pin}",
# Extra requirements for loading our test dags.
"apache-airflow[snowflake]>=2.0.2",
# Connexion's new version breaks Airflow:
# See https://github.com/apache/airflow/issues/35234.
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not a big deal but the period messes up the link on github

# TODO: We should transition to using Airflow's constraints file.
"connexion<3",
# https://github.com/snowflakedb/snowflake-sqlalchemy/issues/350
# Eventually we want to set this to "snowflake-sqlalchemy>=1.4.3".
# However, that doesn't work with older versions of Airflow. Instead
Expand Down
12 changes: 11 additions & 1 deletion metadata-ingestion/src/datahub/ingestion/api/decorators.py
View file
Open in desktop
Expand Up @@ -93,10 +93,20 @@ def capability(
"""

def wrapper(cls: Type) -> Type:
if not hasattr(cls, "__capabilities"):
if not hasattr(cls, "__capabilities") or any(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we just use single underscore instead of double underscore here? I think that's why we have to do this in the first place: https://www.scaler.com/topics/double-underscore-python/

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in this case, we actually do want double underscore to signal that it's specific to that class and not subclasses, since the inheritance piece of it we do ourselves using copies to avoid mutating the same underlying object

also the double underscores don't actually get mangled here because we're not within a class context

# It's from this class and not a superclass.
cls.__capabilities is getattr(base, "__capabilities", None)
for base in cls.__bases__
):
cls.__capabilities = {}
cls.get_capabilities = lambda: cls.__capabilities.values()

# If the superclasses have capability annotations, copy those over.
for base in cls.__bases__:
base_caps = getattr(base, "__capabilities", None)
if base_caps:
cls.__capabilities.update(base_caps)

cls.__capabilities[capability_name] = CapabilitySetting(
capability=capability_name, description=description, supported=supported
)
Expand Down
8 changes: 7 additions & 1 deletion metadata-ingestion/src/datahub/ingestion/source/state/stateful_ingestion_base.py
View file
Open in desktop
Expand Up @@ -15,11 +15,12 @@
from datahub.configuration.time_window_config import BaseTimeWindowConfig
from datahub.configuration.validate_field_rename import pydantic_renamed_field
from datahub.ingestion.api.common import PipelineContext
from datahub.ingestion.api.decorators import capability
from datahub.ingestion.api.ingestion_job_checkpointing_provider_base import (
IngestionCheckpointingProviderBase,
JobId,
)
from datahub.ingestion.api.source import Source, SourceReport
from datahub.ingestion.api.source import Source, SourceCapability, SourceReport
from datahub.ingestion.source.state.checkpoint import Checkpoint, StateType
from datahub.ingestion.source.state.use_case_handler import (
StatefulIngestionUsecaseHandlerBase,
Expand Down Expand Up @@ -177,6 +178,11 @@ class StatefulIngestionReport(SourceReport):
pass


@capability(
SourceCapability.DELETION_DETECTION,
"Optionally enabled via `stateful_ingestion.remove_stale_metadata`",
supported=True,
)
class StatefulIngestionSourceBase(Source):
"""
Defines the base class for all stateful sources.
Expand Down