Repository: apache/airflow
File: providers/google/src/airflow/providers/google/cloud/hooks/bigtable.py (Python)

Code before:

    @GoogleBaseHook.fallback_to_default_project_id
    def delete_table(self, instance_id: str, table_id: str, project_id: str) -> None:
        """
        Delete the specified table in Cloud Bigtable.

        Raises google.api_core.exceptions.NotFound if the table does not exist.

        :param instance_id: The ID of the Cloud Bigtable instance.
        :param table_id: The ID of the table in Cloud Bigtable.
        :param project_id: Optional, Google Cloud project ID where the
            BigTable exists. If set to None or missing,
            the default project_id from the Google Cloud connection is used.
        """
        instance = self.get_instance(instance_id=instance_id, project_id=project_id)
        if instance is None:
            raise RuntimeError(f"Instance {instance_id} did not exist; unable to delete table {table_id}")

        table = instance.table(table_id=table_id)
        try:
            table.delete()
        except google.api_core.exceptions.NotFound:
            self.log.info("The table '%s' no longer exists. Consider it as deleted", table_id)

    @staticmethod
    def update_cluster(instance: Instance, instance_id: str, cluster_id: str, nodes: int) -> None:
        """
        Update number of nodes in the specified Cloud Bigtable cluster.

        Raises google.api_core.exceptions.NotFound if the cluster does not exist.

        :param instance: The Cloud Bigtable instance that owns the cluster.
        :param cluster_id: The ID of the cluster.
        :param nodes: The desired number of nodes.
        """
        cluster = Cluster(cluster_id, instance)
        # "reload" is required to set location_id attribute on cluster.
        try:
            cluster.reload()
        except google.api_core.exceptions.NotFound:
            raise AirflowException(
                f"Dependency: cluster '{cluster_id}' does not exist for instance '{instance_id}'."
            )
        cluster.serve_nodes = nodes
        cluster.update()

    @staticmethod
    def get_column_families_for_table(instance: Instance, table_id: str) -> dict[str, ColumnFamily]:
        """
        Fetch Column Families for the specified table in Cloud Bigtable.


Review comment (suggestion):
Adding a required parameter here is a breaking change.
You should be able to fetch the id from the `instance` parameter.

Code after:
    @GoogleBaseHook.fallback_to_default_project_id
    def delete_table(self, instance_id: str, table_id: str, project_id: str) -> None:
        """
        Delete the specified table in Cloud Bigtable.

        If the instance does not exist, raises RuntimeError.
        If the table does not exist, logs a warning message and returns.

        :param instance_id: The ID of the Cloud Bigtable instance.
        :param table_id: The ID of the table in Cloud Bigtable.
        :param project_id: Optional, Google Cloud project ID where the
            BigTable exists. If set to None or missing,
            the default project_id from the Google Cloud connection is used.
        """
        instance = self.get_instance(instance_id=instance_id, project_id=project_id)
        if instance is None:
            raise RuntimeError(f"Instance {instance_id} did not exist; unable to delete table {table_id}")

        table = instance.table(table_id=table_id)
        try:
            table.delete()
        except google.api_core.exceptions.NotFound:
            self.log.info("The table '%s' no longer exists. Consider it as deleted", table_id)

    @staticmethod
    def update_cluster(instance: Instance, cluster_id: str, nodes: int) -> None:
        """
        Update number of nodes in the specified Cloud Bigtable cluster.

        If the cluster does not exist, raises AirflowException.

        :param instance: The Cloud Bigtable instance that owns the cluster.
        :param cluster_id: The ID of the cluster.
        :param nodes: The desired number of nodes.
        """
        cluster = Cluster(cluster_id, instance)
        # "reload" is required to set location_id attribute on cluster.
        try:
            cluster.reload()
        except google.api_core.exceptions.NotFound:
            raise AirflowException(
                f"Dependency: cluster '{cluster_id}' does not exist for instance '{instance.instance_id}'."
            )
        cluster.serve_nodes = nodes
        cluster.update()

    @staticmethod
    def get_column_families_for_table(instance: Instance, table_id: str) -> dict[str, ColumnFamily]:
        """
        Fetch Column Families for the specified table in Cloud Bigtable.