Subclassing Components

import dagster as dg
from dagster_fivetran import FivetranAccountComponent


class CustomFivetranComponent(FivetranAccountComponent):
    def get_asset_spec(self, data) -> dg.AssetSpec:
        spec = super().get_asset_spec(data)
        return spec.replace_attributes(
            group_name="warehouse_ingestion",
            kinds={"fivetran", "snowflake"},
        )

import dagster as dg
from dagster_fivetran import FivetranAccountComponent


class CustomFivetranComponent(FivetranAccountComponent):
    def execute(self, context: dg.AssetExecutionContext, resource):
        context.log.info("Starting custom pre-sync validation")
        # ... custom logic ...
        yield from super().execute(context, resource)
        context.log.info("Post-sync complete")

from pathlib import Path


class CustomComponent(ExistingStateBackedComponent):
    def write_state_to_path(self, state_path: Path) -> None:
        state = ...  # note: you may need to inspect the parent class to determine the specific state format for `write_state_to_path()`
        state_path.write_text(dg.serialize_value(state))

import dagster as dg


class CustomComponent(ExistingStateBackedComponent):
    def build_defs_from_state(self, context, state_path: Path) -> dg.Definitions:
        base_defs = super().build_defs_from_state(context, state_path)
        # Add a custom sensor alongside the base definitions
        return dg.Definitions(
            assets=base_defs.assets,
            sensors=[my_custom_sensor],
            resources=base_defs.resources,
        )