From 0b668b6e5ee857c603aa9ca07d9aa2273010891c Mon Sep 17 00:00:00 2001
From: Heorhi Parkhomenka <Heorhi_Parkhomenka@epam.com>
Date: Tue, 25 Oct 2022 13:34:07 +0200
Subject: [PATCH 1/4] Add operators, links, update system tests and docs

---
 .../providers/google/cloud/hooks/dataform.py  | 378 ++++++++-
 .../providers/google/cloud/links/dataform.py  |  66 ++
 .../google/cloud/operators/dataform.py        | 775 +++++++++++++++++-
 .../providers/google/cloud/utils/dataform.py  | 213 +++++
 airflow/providers/google/provider.yaml        |   2 +
 .../operators/cloud/dataform.rst              |  84 ++
 .../google/cloud/hooks/test_dataform.py       | 202 +++++
 .../google/cloud/operators/test_dataform.py   | 239 ++++++
 .../google/cloud/dataform/example_dataform.py | 149 +++-
 tests/utils/test_dataform.py                  |  34 +
 10 files changed, 2110 insertions(+), 32 deletions(-)
 create mode 100644 airflow/providers/google/cloud/utils/dataform.py
 create mode 100644 tests/utils/test_dataform.py

diff --git a/airflow/providers/google/cloud/hooks/dataform.py b/airflow/providers/google/cloud/hooks/dataform.py
index e078b99665eb7..c9c4070309f8b 100644
--- a/airflow/providers/google/cloud/hooks/dataform.py
+++ b/airflow/providers/google/cloud/hooks/dataform.py
@@ -22,7 +22,14 @@
 from google.api_core.gapic_v1.method import DEFAULT, _MethodDefault
 from google.api_core.retry import Retry
 from google.cloud.dataform_v1beta1 import DataformClient
-from google.cloud.dataform_v1beta1.types import CompilationResult, WorkflowInvocation
+from google.cloud.dataform_v1beta1.types import (
+    CompilationResult,
+    InstallNpmPackagesResponse,
+    Repository,
+    WorkflowInvocation,
+    Workspace,
+    WriteFileResponse,
+)
 
 from airflow import AirflowException
 from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
@@ -249,3 +256,372 @@ def cancel_workflow_invocation(
         client.cancel_workflow_invocation(
             request={"name": name}, retry=retry, timeout=timeout, metadata=metadata
         )
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def create_repository(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ) -> Repository:
+        """
+        Creates repository
+
+        :param project_id: Required. The ID of the Google Cloud project where repository should be.
+        :param region: Required. The ID of the Google Cloud region where repository should be.
+        :param repository_id: Required. The ID of the new Dataform repository.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        parent = f"projects/{project_id}/locations/{region}"
+        request = {
+            "parent": parent,
+            "repository_id": repository_id,
+        }
+
+        repository = client.create_repository(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        return repository
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def delete_repository(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        force: bool = True,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ) -> None:
+        """
+        Deletes repository.
+
+        :param project_id: Required. The ID of the Google Cloud project where repository located.
+        :param region: Required. The ID of the Google Cloud region where repository located.
+        :param repository_id: Required. The ID of the Dataform repository that should be deleted.
+        :param force: If set to true, any child resources of this repository will also be deleted.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        name = f"projects/{project_id}/locations/{region}/repositories/{repository_id}"
+        request = {
+            'name': name,
+            'force': force,
+        }
+
+        client.delete_repository(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def create_workspace(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ) -> Workspace:
+        """
+        Creates workspace.
+
+        :param project_id: Required. The ID of the Google Cloud project where workspace should be.
+        :param region: Required. The ID of the Google Cloud region where workspace should be.
+        :param repository_id: Required. The ID of the Dataform repository where workspace should be.
+        :param workspace_id: Required. The ID of the new Dataform workspace.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        parent = f"projects/{project_id}/locations/{region}/repositories/{repository_id}"
+
+        request = {'parent': parent, 'workspace_id': workspace_id}
+
+        workspace = client.create_workspace(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        return workspace
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def delete_workspace(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ):
+        """
+        Deletes workspace.
+
+        :param project_id: Required. The ID of the Google Cloud project where workspace located.
+        :param region: Required. The ID of the Google Cloud region where workspace located.
+        :param repository_id: Required. The ID of the Dataform repository where workspace located.
+        :param workspace_id: Required. The ID of the Dataform workspace that should be deleted.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        workspace_path = (
+            f"projects/{project_id}/locations/{region}/"
+            f"repositories/{repository_id}/workspaces/{workspace_id}"
+        )
+        request = {
+            "name": workspace_path,
+        }
+
+        client.delete_workspace(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def write_file(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        filepath: str,
+        contents: bytes,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ) -> WriteFileResponse:
+        """
+        Writes a new file to the specified workspace.
+
+        :param project_id: Required. The ID of the Google Cloud project where workspace located.
+        :param region: Required. The ID of the Google Cloud region where workspace located.
+        :param repository_id: Required. The ID of the Dataform repository where workspace located.
+        :param workspace_id: Required. The ID of the Dataform workspace where files should be created.
+        :param filepath: Required. Path to file including name of the file relative to workspace root.
+        :param contents: Required. Content of the file to be writed.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        workspace_path = (
+            f"projects/{project_id}/locations/{region}/"
+            f"repositories/{repository_id}/workspaces/{workspace_id}"
+        )
+        request = {
+            "workspace": workspace_path,
+            "path": filepath,
+            "contents": contents,
+        }
+
+        response = client.write_file(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        return response
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def make_directory(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        path: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ) -> dict:
+        """
+        Makes new directory in specified workspace.
+
+        :param project_id: Required. The ID of the Google Cloud project where workspace located.
+        :param region: Required. The ID of the Google Cloud region where workspace located.
+        :param repository_id: Required. The ID of the Dataform repository where workspace located.
+        :param workspace_id: Required. The ID of the Dataform workspace where directory should be created.
+        :param path: Required. The directory's full path including new directory name,
+            relative to the workspace root.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        workspace_path = (
+            f"projects/{project_id}/locations/{region}/"
+            f"repositories/{repository_id}/workspaces/{workspace_id}"
+        )
+        request = {
+            "workspace": workspace_path,
+            "path": path,
+        }
+
+        response = client.make_directory(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        return response
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def remove_directory(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        path: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ):
+        """
+        Removes directory in specified workspace.
+
+        :param project_id: Required. The ID of the Google Cloud project where workspace located.
+        :param region: Required. The ID of the Google Cloud region where workspace located.
+        :param repository_id: Required. The ID of the Dataform repository where workspace located.
+        :param workspace_id: Required. The ID of the Dataform workspace where directory located.
+        :param path: Required. The directory's full path including directory name,
+            relative to the workspace root.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        workspace_path = (
+            f"projects/{project_id}/locations/{region}/"
+            f"repositories/{repository_id}/workspaces/{workspace_id}"
+        )
+        request = {
+            "workspace": workspace_path,
+            "path": path,
+        }
+
+        client.remove_directory(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def remove_file(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        filepath: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ):
+        """
+        Removes file in specified workspace.
+
+        :param project_id: Required. The ID of the Google Cloud project where workspace located.
+        :param region: Required. The ID of the Google Cloud region where workspace located.
+        :param repository_id: Required. The ID of the Dataform repository where workspace located.
+        :param workspace_id: Required. The ID of the Dataform workspace where directory located.
+        :param filepath: Required. The full path including name of the file, relative to the workspace root.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        workspace_path = (
+            f"projects/{project_id}/locations/{region}/"
+            f"repositories/{repository_id}/workspaces/{workspace_id}"
+        )
+        request = {
+            "workspace": workspace_path,
+            "path": filepath,
+        }
+
+        client.remove_file(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+    @GoogleBaseHook.fallback_to_default_project_id
+    def install_npm_packages(
+        self,
+        *,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+    ) -> InstallNpmPackagesResponse:
+        """
+        Installs npm dependencies in the provided workspace. Requires "package.json"
+        to be created in workspace
+
+        :param project_id: Required. The ID of the Google Cloud project where workspace located.
+        :param region: Required. The ID of the Google Cloud region where workspace located.
+        :param repository_id: Required. The ID of the Dataform repository where workspace located.
+        :param workspace_id: Required. The ID of the Dataform workspace.
+        :param retry: Designation of what errors, if any, should be retried.
+        :param timeout: The timeout for this request.
+        :param metadata: Strings which should be sent along with the request as metadata.
+        """
+        client = self.get_dataform_client()
+        workspace_path = (
+            f"projects/{project_id}/locations/{region}/"
+            f"repositories/{repository_id}/workspaces/{workspace_id}"
+        )
+        request = {
+            "workspace": workspace_path,
+        }
+
+        response = client.install_npm_packages(
+            request=request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        return response
diff --git a/airflow/providers/google/cloud/links/dataform.py b/airflow/providers/google/cloud/links/dataform.py
index 1faf6ec1db0bc..53dcbed11b835 100644
--- a/airflow/providers/google/cloud/links/dataform.py
+++ b/airflow/providers/google/cloud/links/dataform.py
@@ -25,12 +25,24 @@
 
 if TYPE_CHECKING:
     from airflow.utils.context import Context
+
 DATAFORM_BASE_LINK = "/bigquery/dataform"
 DATAFORM_WORKFLOW_INVOCATION_LINK = (
     DATAFORM_BASE_LINK
     + "/locations/{region}/repositories/{repository_id}/workflows/"
     + "{workflow_invocation_id}?project={project_id}"
 )
+DATAFORM_REPOSITORY_LINK = (
+    DATAFORM_BASE_LINK
+    + "/locations/{region}/repositories/{repository_id}/"
+    + "details/workspaces?project={project_id}"
+)
+DATAFORM_WORKSPACE_LINK = (
+    DATAFORM_BASE_LINK
+    + "/locations/{region}/repositories/{repository_id}/"
+    + "workspaces/{workspace_id}/"
+    + "files/?project={project_id}"
+)
 
 
 class DataformWorkflowInvocationLink(BaseGoogleLink):
@@ -59,3 +71,57 @@ def persist(
                 "workflow_invocation_id": workflow_invocation_id,
             },
         )
+
+
+class DataformRepositoryLink(BaseGoogleLink):
+    """Helper class for constructing Dataflow repository link."""
+
+    name = "Dataform Repository"
+    key = "dataform_repository"
+    format_str = DATAFORM_REPOSITORY_LINK
+
+    @staticmethod
+    def persist(
+        operator_instance: BaseOperator,
+        context: Context,
+        project_id: str,
+        region: str,
+        repository_id: str,
+    ) -> None:
+        operator_instance.xcom_push(
+            context=context,
+            key=DataformRepositoryLink.key,
+            value={
+                "project_id": project_id,
+                "region": region,
+                "repository_id": repository_id,
+            },
+        )
+
+
+class DataformWorkspaceLink(BaseGoogleLink):
+    """Helper class for constructing Datafrom workspace link."""
+
+    name = "Dataform Workspace"
+    key = "dataform_workspace"
+    format_str = DATAFORM_WORKSPACE_LINK
+
+    @staticmethod
+    def persist(
+        operator_instance: BaseOperator,
+        context: Context,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+    ) -> None:
+        operator_instance.xcom_push(
+            context=context,
+            key=DataformWorkspaceLink.key,
+            value={
+                "project_id": project_id,
+                "region": region,
+                "repository_id": repository_id,
+                "workspace_id": workspace_id,
+            },
+        )
diff --git a/airflow/providers/google/cloud/operators/dataform.py b/airflow/providers/google/cloud/operators/dataform.py
index 8d3eecd34bfa6..d6801031596bf 100644
--- a/airflow/providers/google/cloud/operators/dataform.py
+++ b/airflow/providers/google/cloud/operators/dataform.py
@@ -18,14 +18,26 @@
 
 from typing import TYPE_CHECKING, Sequence
 
-from airflow.providers.google.cloud.links.dataform import DataformWorkflowInvocationLink
+from airflow.providers.google.cloud.links.dataform import (
+    DataformRepositoryLink,
+    DataformWorkflowInvocationLink,
+    DataformWorkspaceLink,
+)
 
 if TYPE_CHECKING:
     from airflow.utils.context import Context
 
 from google.api_core.gapic_v1.method import DEFAULT, _MethodDefault
 from google.api_core.retry import Retry
-from google.cloud.dataform_v1beta1.types import CompilationResult, WorkflowInvocation
+from google.cloud.dataform_v1beta1.types import (
+    CompilationResult,
+    InstallNpmPackagesResponse,
+    MakeDirectoryResponse,
+    Repository,
+    WorkflowInvocation,
+    Workspace,
+    WriteFileResponse,
+)
 
 from airflow.models import BaseOperator
 from airflow.providers.google.cloud.hooks.dataform import DataformHook
@@ -410,3 +422,762 @@ def execute(self, context: Context):
             timeout=self.timeout,
             metadata=self.metadata,
         )
+
+
+class DataformCreateRepositoryOperator(BaseOperator):
+    """
+    Creates repository.
+
+    :param project_id: Required. The ID of the Google Cloud project that the task belongs to.
+    :param region: Required. The ID of the Google Cloud region that the task belongs to.
+    :param repository_id: Required. The ID of the Dataform repository that the task belongs to.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    operator_extra_links = (DataformRepositoryLink(),)
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.project_id = project_id
+        self.region = region
+        self.repository_id = repository_id
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> dict:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+        repository = hook.create_repository(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+
+        DataformRepositoryLink.persist(
+            operator_instance=self,
+            context=context,
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+        )
+
+        return Repository.to_dict(repository)
+
+
+class DataformDeleteRepositoryOperator(BaseOperator):
+    """
+    Deletes repository.
+
+    :param project_id: Required. The ID of the Google Cloud project where repository located.
+    :param region: Required. The ID of the Google Cloud region where repository located.
+    :param repository_id: Required. The ID of the Dataform repository that should be deleted.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        force: bool = True,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.repository_id = repository_id
+        self.project_id = project_id
+        self.region = region
+        self.force = force
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> None:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+        hook.delete_repository(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            force=self.force,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+
+
+class DataformCreateWorkspaceOperator(BaseOperator):
+    """
+    Creates workspace.
+
+    :param project_id: Required. The ID of the Google Cloud project where workspace should be in.
+    :param region: Required. Name of the Google Cloud region that where workspace should be in.
+    :param repository_id: Required. The ID of the Dataform repository that the workspace belongs to.
+    :param workspace_id: Required. The ID of the new workspace that will be created.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    operator_extra_links = (DataformWorkspaceLink(),)
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.project_id = project_id
+        self.workspace_id = workspace_id
+        self.repository_id = repository_id
+        self.region = region
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> dict:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+        workspace = hook.create_workspace(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            workspace_id=self.workspace_id,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+
+        DataformWorkspaceLink.persist(
+            operator_instance=self,
+            context=context,
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            workspace_id=self.workspace_id,
+        )
+
+        return Workspace.to_dict(workspace)
+
+
+class DataformDeleteWorkspaceOperator(BaseOperator):
+    """
+    Deletes workspace.
+
+    :param project_id: Required. The ID of the Google Cloud project where workspace located.
+    :param region: Required. The ID of the Google Cloud region where workspace located.
+    :param repository_id: Required. The ID of the Dataform repository where workspace located.
+    :param workspace_id: Required. The ID of the Dataform workspace that should be deleted.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "workspace_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.project_id = project_id
+        self.region = region
+        self.repository_id = repository_id
+        self.workspace_id = workspace_id
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> None:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+        hook.delete_workspace(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            workspace_id=self.workspace_id,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+
+
+class DataformWriteFileOperator(BaseOperator):
+    """
+    Writes new file to specified workspace.
+
+    :param project_id: Required. The ID of the Google Cloud project where workspace located.
+    :param region: Required. The ID of the Google Cloud region where workspace located.
+    :param repository_id: Required. The ID of the Dataform repository where workspace located.
+    :param workspace_id: Required. The ID of the Dataform workspace where files should be created.
+    :param filepath: Required. Path to file including name of the file relative to workspace root.
+    :param contents: Required. Content of the file to be writed.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "workspace_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        filepath: str,
+        contents: bytes,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.project_id = project_id
+        self.region = region
+        self.repository_id = repository_id
+        self.workspace_id = workspace_id
+        self.filepath = filepath
+        self.contents = contents
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> dict:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+        write_file_response = hook.write_file(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            workspace_id=self.workspace_id,
+            filepath=self.filepath,
+            contents=self.contents,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+        return WriteFileResponse.to_dict(write_file_response)
+
+
+class DataformMakeDirectoryOperator(BaseOperator):
+    """
+    Makes new directory in specified workspace.
+
+    :param project_id: Required. The ID of the Google Cloud project where workspace located.
+    :param region: Required. The ID of the Google Cloud region where workspace located.
+    :param repository_id: Required. The ID of the Dataform repository where workspace located.
+    :param workspace_id: Required. The ID of the Dataform workspace where directory should be created.
+    :param path: Required. The directory's full path including directory name, relative to the workspace root.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "workspace_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        directory_path: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.project_id = project_id
+        self.region = region
+        self.repository_id = repository_id
+        self.workspace_id = workspace_id
+        self.directory_path = directory_path
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> dict:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+        make_directory_response = hook.make_directory(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            workspace_id=self.workspace_id,
+            path=self.directory_path,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+
+        return MakeDirectoryResponse.to_dict(make_directory_response)
+
+
+class DataformRemoveFileOperator(BaseOperator):
+    """
+    Removes file in specified workspace.
+
+    :param project_id: Required. The ID of the Google Cloud project where workspace located.
+    :param region: Required. The ID of the Google Cloud region where workspace located.
+    :param repository_id: Required. The ID of the Dataform repository where workspace located.
+    :param workspace_id: Required. The ID of the Dataform workspace where directory located.
+    :param filepath: Required. The full path including name of the file, relative to the workspace root.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "workspace_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        filepath: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.project_id = project_id
+        self.region = region
+        self.repository_id = repository_id
+        self.workspace_id = workspace_id
+        self.filepath = filepath
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> None:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+        hook.remove_file(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            workspace_id=self.workspace_id,
+            filepath=self.filepath,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+
+
+class DataformRemoveDirectoryOperator(BaseOperator):
+    """
+    Removes directory in specified workspace.
+
+    :param project_id: Required. The ID of the Google Cloud project where workspace located.
+    :param region: Required. The ID of the Google Cloud region where workspace located.
+    :param repository_id: Required. The ID of the Dataform repository where workspace located.
+    :param workspace_id: Required. The ID of the Dataform workspace where directory located.
+    :param path: Required. The directory's full path including directory name, relative to the workspace root.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "workspace_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        directory_path: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.project_id = project_id
+        self.region = region
+        self.repository_id = repository_id
+        self.workspace_id = workspace_id
+        self.directory_path = directory_path
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> None:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+        hook.remove_directory(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            workspace_id=self.workspace_id,
+            path=self.directory_path,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+
+
+class DataformInstallNpmPackagesOperator(BaseOperator):
+    """
+    Installs npm dependencies in the provided workspace. Requires "package.json" to be created in workspace
+
+    :param project_id: Required. The ID of the Google Cloud project where workspace located.
+    :param region: Required. The ID of the Google Cloud region where workspace located.
+    :param repository_id: Required. The ID of the Dataform repository where workspace located.
+    :param workspace_id: Required. The ID of the Dataform workspace.
+    :param retry: Designation of what errors, if any, should be retried.
+    :param timeout: The timeout for this request.
+    :param metadata: Strings which should be sent along with the request as metadata.
+    :param gcp_conn_id: The connection ID to use when fetching connection info.
+    :param delegate_to: The account to impersonate, if any. For this to work, the service accountmaking the
+        request must have  domain-wide delegation enabled.
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    """
+
+    template_fields = (
+        "project_id",
+        "repository_id",
+        "workspace_id",
+        "delegate_to",
+        "impersonation_chain",
+    )
+
+    def __init__(
+        self,
+        project_id: str,
+        region: str,
+        repository_id: str,
+        workspace_id: str,
+        retry: Retry | _MethodDefault = DEFAULT,
+        timeout: float | None = None,
+        metadata: Sequence[tuple[str, str]] = (),
+        gcp_conn_id: str = "google_cloud_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+        *args,
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+
+        self.project_id = project_id
+        self.region = region
+        self.repository_id = repository_id
+        self.workspace_id = workspace_id
+
+        self.retry = retry
+        self.timeout = timeout
+        self.metadata = metadata
+
+        self.gcp_conn_id = gcp_conn_id
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context: Context) -> dict:
+        hook = DataformHook(
+            gcp_conn_id=self.gcp_conn_id,
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+
+        response = hook.install_npm_packages(
+            project_id=self.project_id,
+            region=self.region,
+            repository_id=self.repository_id,
+            workspace_id=self.workspace_id,
+            retry=self.retry,
+            timeout=self.timeout,
+            metadata=self.metadata,
+        )
+
+        return InstallNpmPackagesResponse.to_dict(response)
diff --git a/airflow/providers/google/cloud/utils/dataform.py b/airflow/providers/google/cloud/utils/dataform.py
new file mode 100644
index 0000000000000..2596f37b0167d
--- /dev/null
+++ b/airflow/providers/google/cloud/utils/dataform.py
@@ -0,0 +1,213 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import json
+from enum import Enum
+from typing import Mapping
+
+from airflow.providers.google.cloud.operators.dataform import (
+    DataformInstallNpmPackagesOperator,
+    DataformMakeDirectoryOperator,
+    DataformWriteFileOperator,
+)
+
+
+class DataformLocations(str, Enum):
+    """Enum for storing available locations for resources in Dataform."""
+
+    US = "US"
+    EUROPE = "EU"
+
+
+def make_initialization_workspace_flow(
+    project_id: str,
+    region: str,
+    repository_id: str,
+    workspace_id: str,
+    package_name: str | None = None,
+) -> tuple:
+    """
+    Creates flow which simulates the initialization of the default project.
+    As a result project with the layout listed below will be created.
+    - definitions/
+      |-first_view.sqlx
+      |-second_view.sqlx
+    - includes/
+    - .gitignore
+    - dataform.json
+    - package.json
+    - package-lock.json
+    :param project_id: Required. The ID of the Google Cloud project where workspace located.
+    :param region: Required. The ID of the Google Cloud region where workspace located.
+    :param repository_id: Required. The ID of the Dataform repository where workspace located.
+    :param workspace_id: Required. The ID of the Dataform workspace which requires initialization.
+    :param package_name: Name of the package. This value will be used in package.json. If value is not
+        provided then workspace_id will be used.
+    """
+    make_definitions_directory = DataformMakeDirectoryOperator(
+        task_id='make-definitions-directory',
+        project_id=project_id,
+        region=region,
+        repository_id=repository_id,
+        workspace_id=workspace_id,
+        directory_path='definitions',
+    )
+
+    first_view_content = b"""
+        -- This is an example SQLX file to help you learn the basics of Dataform.
+        -- Visit https://cloud.google.com/dataform/docs/how-to for more information on how to configure
+            your SQL workflow.
+
+        -- You can delete this file, then commit and push your changes to your repository when you are ready.
+
+        -- Config blocks allow you to configure, document, and test your data assets.
+        config {
+          type: "view", // Creates a view in BigQuery. Try changing to "table" instead.
+          columns: {
+            test: "A description for the test column", // Column descriptions are pushed to BigQuery.
+          }
+        }
+
+        -- The rest of a SQLX file contains your SELECT statement used to create the table.
+
+        SELECT 1 as test
+    """
+    make_first_view_file = DataformWriteFileOperator(
+        task_id='write-first-view',
+        project_id=project_id,
+        region=region,
+        repository_id=repository_id,
+        workspace_id=workspace_id,
+        filepath='definitions/first_view.sqlx',
+        contents=first_view_content,
+    )
+
+    second_view_content = b"""
+        config { type: "view" }
+
+        -- Use the ref() function to manage dependencies.
+        -- Learn more about ref() and other built in functions
+            here: https://cloud.google.com/dataform/docs/dataform-core
+
+        SELECT test from ${ref("first_view")}
+    """
+    make_second_view_file = DataformWriteFileOperator(
+        task_id="write-second-view",
+        project_id=project_id,
+        region=region,
+        repository_id=repository_id,
+        workspace_id=workspace_id,
+        filepath='definitions/second_view.sqlx',
+        contents=second_view_content,
+    )
+
+    make_includes_directory = DataformMakeDirectoryOperator(
+        task_id="make-includes-directory",
+        project_id=project_id,
+        region=region,
+        repository_id=repository_id,
+        workspace_id=workspace_id,
+        directory_path="includes",
+    )
+
+    gitignore_contents = b"""
+    node_modules/
+    """
+    make_gitignore_file = DataformWriteFileOperator(
+        task_id="write-gitignore-file",
+        project_id=project_id,
+        region=region,
+        repository_id=repository_id,
+        workspace_id=workspace_id,
+        filepath=".gitignore",
+        contents=gitignore_contents,
+    )
+
+    default_location: str = define_default_location(region).value
+    dataform_config_content = json.dumps(
+        {
+            "defaultSchema": "dataform",
+            "assertionSchema": "dataform_assertions",
+            "warehouse": "bigquery",
+            "defaultDatabase": project_id,
+            "defaultLocation": default_location,
+        },
+        indent=4,
+    ).encode()
+    make_dataform_config_file = DataformWriteFileOperator(
+        task_id="write-dataform-config-file",
+        project_id=project_id,
+        region=region,
+        repository_id=repository_id,
+        workspace_id=workspace_id,
+        filepath="dataform.json",
+        contents=dataform_config_content,
+    )
+
+    package_name = package_name if package_name else workspace_id
+    package_json_content = json.dumps(
+        {
+            "name": package_name,
+            "dependencies": {
+                "@dataform/core": "2.0.1",
+            },
+        },
+        indent=4,
+    ).encode()
+    make_package_json_file = DataformWriteFileOperator(
+        task_id="write-package-json",
+        project_id=project_id,
+        region=region,
+        repository_id=repository_id,
+        workspace_id=workspace_id,
+        filepath="package.json",
+        contents=package_json_content,
+    )
+
+    install_npm_packages = DataformInstallNpmPackagesOperator(
+        task_id="install-npm-packages",
+        project_id=project_id,
+        region=region,
+        repository_id=repository_id,
+        workspace_id=workspace_id,
+    )
+
+    (
+        make_definitions_directory
+        >> make_first_view_file
+        >> make_second_view_file
+        >> make_gitignore_file
+        >> make_dataform_config_file
+        >> make_package_json_file
+        >> install_npm_packages
+        >> make_includes_directory
+    )
+
+    return make_definitions_directory, make_includes_directory
+
+
+def define_default_location(region: str) -> DataformLocations:
+    if "us" in region:
+        return DataformLocations.US
+    elif "europe" in region:
+        return DataformLocations.EUROPE
+
+    regions_mapping: Mapping[str, DataformLocations] = {}
+
+    return regions_mapping[region]
diff --git a/airflow/providers/google/provider.yaml b/airflow/providers/google/provider.yaml
index 268fc32bc9c57..c4182684d89c8 100644
--- a/airflow/providers/google/provider.yaml
+++ b/airflow/providers/google/provider.yaml
@@ -964,6 +964,8 @@ extra-links:
   - airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleLink
   - airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink
   - airflow.providers.google.cloud.operators.mlengine.AIPlatformConsoleLink
+  - airflow.providers.google.cloud.links.dataform.DataformRepositoryLink
+  - airflow.providers.google.cloud.links.dataform.DataformWorkspaceLink
   - airflow.providers.google.cloud.links.dataform.DataformWorkflowInvocationLink
   - airflow.providers.google.cloud.operators.datafusion.DataFusionInstanceLink
   - airflow.providers.google.cloud.operators.datafusion.DataFusionPipelineLink
diff --git a/docs/apache-airflow-providers-google/operators/cloud/dataform.rst b/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
index de9e5871a0a91..ffb7505c80243 100644
--- a/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
+++ b/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
@@ -34,6 +34,30 @@ Configuration
 Before you can use the Dataform operators you need to initialize repository and workspace, for more information
 about this visit `Dataform Production documentation <Product documentation <https://cloud.google.com/dataform/docs/reference>`__
 
+Create Repository
+-----------------
+Create repository for tracking your code in Dataform service. Example of usage can be seen below:
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformCreateRepositoryOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 0
+    :start-after: [START howto_operator_create_repository]
+    :end-before: [END howto_operator_create_repository]
+
+Create Workspace
+----------------
+Create workspace for storing your code in Dataform service. Example of usage can be seen below:
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformCreateWorkspaceOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 0
+    :start-after: [START howto_operator_create_workspace]
+    :end-before: [END howto_operator_create_workspace]
+
 
 Create Compilation Result
 -------------------------
@@ -108,3 +132,63 @@ To cancel a Workflow Invocation you can use:
     :dedent: 4
     :start-after: [START howto_operator_cancel_workflow_invocation]
     :end-before: [END howto_operator_cancel_workflow_invocation]
+
+Delete Repository
+-----------------
+Deletes repository. Example of usage can be seen below:
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformDeleteRepositoryOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 0
+    :start-after: [START howto_operator_delete_workspace]
+    :end-before: [END howto_operator_delete_workspace]
+
+Delete Workspace
+----------------
+Deletes workspace. Example of usage can be seen below:
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformDeleteRepositoryOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 0
+    :start-after: [START howto_operator_delete_repository]
+    :end-before: [END howto_operator_create_workspace]
+
+Remove file
+-----------
+Removes file. Example of usage can be seen below:
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformRemoveFileOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 0
+    :start-after: [START howto_operator_remove_file]
+    :end-before: [END howto_operator_remove_file]
+
+Remove directory
+----------------
+Removes directory. Example of usage can be seen below:
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformRemoveDirectoryOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 0
+    :start-after: [START howto_operator_remove_directory]
+    :end-before: [END howto_operator_remove_directory]
+
+Initialize workspace
+--------------------
+Creates default projects structure for provided workspace. Before it can be done workspace and repository should be created. Example of usage can be seen below:
+
+:class:`~airflow.providers.google.cloud.utils.dataform.make_initialization_workspace_flow`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 0
+    :start-after: [START howto_initialize_workspace]
+    :end-before: [END howto_initialize_workspace]
diff --git a/tests/providers/google/cloud/hooks/test_dataform.py b/tests/providers/google/cloud/hooks/test_dataform.py
index a0bbd664e8188..07bc89110f807 100644
--- a/tests/providers/google/cloud/hooks/test_dataform.py
+++ b/tests/providers/google/cloud/hooks/test_dataform.py
@@ -48,6 +48,9 @@
     ),
 }
 WORKFLOW_INVOCATION_ID = "test_workflow_invocation_id"
+PATH_TO_FOLDER = "path/to/folder"
+FILEPATH = "path/to/file.txt"
+FILE_CONTENTS = b"test content"
 
 
 class TestDataflowHook(unittest.TestCase):
@@ -155,3 +158,202 @@ def test_cancel_workflow_invocation(self, mock_client):
             timeout=None,
             metadata=(),
         )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_create_repository(self, mock_client):
+        self.hook.create_repository(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+        )
+        parent = f"projects/{PROJECT_ID}/locations/{REGION}"
+        mock_client.return_value.create_repository.assert_called_once_with(
+            request={"parent": parent, "repository_id": REPOSITORY_ID},
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_create_workspace(self, mock_client):
+        self.hook.create_workspace(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+        )
+        parent = f"projects/{PROJECT_ID}/locations/{REGION}/repositories/{REPOSITORY_ID}"
+
+        mock_client.return_value.create_workspace.assert_called_once_with(
+            request={
+                'parent': parent,
+                'workspace_id': WORKSPACE_ID,
+            },
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_delete_repository(self, mock_client):
+        force = True
+        self.hook.delete_repository(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            force=force,
+        )
+        name = f"projects/{PROJECT_ID}/locations/{REGION}/repositories/{REPOSITORY_ID}"
+
+        mock_client.return_value.delete_repository.assert_called_once_with(
+            request={
+                "name": name,
+                "force": force,
+            },
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_delete_workspace(self, mock_client):
+        self.hook.delete_workspace(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+        )
+        name = (
+            f"projects/{PROJECT_ID}/locations/{REGION}/"
+            f"repositories/{REPOSITORY_ID}/workspaces/{WORKSPACE_ID}"
+        )
+
+        mock_client.return_value.delete_workspace.assert_called_once_with(
+            request={
+                "name": name,
+            },
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_write_file(self, mock_client):
+        self.hook.write_file(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            filepath=FILEPATH,
+            contents=FILE_CONTENTS,
+        )
+        workspace_path = (
+            f"projects/{PROJECT_ID}/locations/{REGION}/"
+            f"repositories/{REPOSITORY_ID}/workspaces/{WORKSPACE_ID}"
+        )
+
+        mock_client.return_value.write_file.assert_called_once_with(
+            request={
+                "workspace": workspace_path,
+                "path": FILEPATH,
+                "contents": FILE_CONTENTS,
+            },
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_make_directory(self, mock_client):
+        self.hook.make_directory(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            path=PATH_TO_FOLDER,
+        )
+        workspace_path = (
+            f"projects/{PROJECT_ID}/locations/{REGION}/"
+            f"repositories/{REPOSITORY_ID}/workspaces/{WORKSPACE_ID}"
+        )
+
+        mock_client.return_value.make_directory.assert_called_once_with(
+            request={
+                "workspace": workspace_path,
+                "path": PATH_TO_FOLDER,
+            },
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_remove_directory(self, mock_client):
+        self.hook.remove_directory(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            path=PATH_TO_FOLDER,
+        )
+        workspace_path = (
+            f"projects/{PROJECT_ID}/locations/{REGION}/"
+            f"repositories/{REPOSITORY_ID}/workspaces/{WORKSPACE_ID}"
+        )
+
+        mock_client.return_value.remove_directory.assert_called_once_with(
+            request={
+                "workspace": workspace_path,
+                "path": PATH_TO_FOLDER,
+            },
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_remove_file(self, mock_client):
+        self.hook.remove_file(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            filepath=FILEPATH,
+        )
+        workspace_path = (
+            f"projects/{PROJECT_ID}/locations/{REGION}/"
+            f"repositories/{REPOSITORY_ID}/workspaces/{WORKSPACE_ID}"
+        )
+
+        mock_client.return_value.remove_file.assert_called_once_with(
+            request={
+                "workspace": workspace_path,
+                "path": FILEPATH,
+            },
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
+    def test_install_npm_packages(self, mock_client):
+        self.hook.install_npm_packages(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+        )
+
+        workspace_path = (
+            f"projects/{PROJECT_ID}/locations/{REGION}/"
+            f"repositories/{REPOSITORY_ID}/workspaces/{WORKSPACE_ID}"
+        )
+
+        mock_client.return_value.install_npm_packages.assert_called_once_with(
+            request={
+                "workspace": workspace_path,
+            },
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
diff --git a/tests/providers/google/cloud/operators/test_dataform.py b/tests/providers/google/cloud/operators/test_dataform.py
index 964487876718d..47a1eecdc1be0 100644
--- a/tests/providers/google/cloud/operators/test_dataform.py
+++ b/tests/providers/google/cloud/operators/test_dataform.py
@@ -24,14 +24,30 @@
 from airflow.providers.google.cloud.operators.dataform import (
     DataformCancelWorkflowInvocationOperator,
     DataformCreateCompilationResultOperator,
+    DataformCreateRepositoryOperator,
     DataformCreateWorkflowInvocationOperator,
+    DataformCreateWorkspaceOperator,
+    DataformDeleteRepositoryOperator,
+    DataformDeleteWorkspaceOperator,
     DataformGetCompilationResultOperator,
     DataformGetWorkflowInvocationOperator,
+    DataformInstallNpmPackagesOperator,
+    DataformMakeDirectoryOperator,
+    DataformRemoveDirectoryOperator,
+    DataformRemoveFileOperator,
+    DataformWriteFileOperator,
 )
 
 HOOK_STR = "airflow.providers.google.cloud.operators.dataform.DataformHook"
 WORKFLOW_INVOCATION_STR = "airflow.providers.google.cloud.operators.dataform.WorkflowInvocation"
 COMPILATION_RESULT_STR = "airflow.providers.google.cloud.operators.dataform.CompilationResult"
+REPOSITORY_STR = "airflow.providers.google.cloud.operators.dataform.Repository"
+WORKSPACE_STR = "airflow.providers.google.cloud.operators.dataform.Workspace"
+WRITE_FILE_RESPONSE_STR = "airflow.providers.google.cloud.operators.dataform.WriteFileResponse"
+MAKE_DIRECTORY_RESPONSE_STR = "airflow.providers.google.cloud.operators.dataform.MakeDirectoryResponse"
+INSTALL_NPM_PACKAGES_RESPONSE_STR = (
+    "airflow.providers.google.cloud.operators.dataform.InstallNpmPackagesResponse"
+)
 
 PROJECT_ID = "project-id"
 REGION = "region"
@@ -42,6 +58,9 @@
 GCP_CONN_ID = "google_cloud_default"
 DELEGATE_TO = "test-delegate-to"
 IMPERSONATION_CHAIN = ["ACCOUNT_1", "ACCOUNT_2", "ACCOUNT_3"]
+FILEPATH = "path/to/file.txt"
+FILE_CONTENT = b"test content"
+DIRECTORY_PATH = "path/to/directory"
 
 WORKFLOW_INVOCATION = {
     "compilation_result": (
@@ -175,3 +194,223 @@ def test_execute(self, hook_mock):
             timeout=None,
             metadata=(),
         )
+
+
+class TestDataformCreateRepositoryOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    @mock.patch(REPOSITORY_STR)
+    def test_execute(self, _, hook_mock):
+        op = DataformCreateRepositoryOperator(
+            task_id="create-repository",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+        )
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.create_repository.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+
+class TestDataformDeleteRepositoryOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    def test_execute(self, hook_mock):
+        force = True
+        op = DataformDeleteRepositoryOperator(
+            task_id="delete-repository",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            force=force,
+        )
+
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.delete_repository.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            force=force,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+
+class TestDataformCreateWorkspaceOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    @mock.patch(WORKSPACE_STR)
+    def test_execute(self, _, hook_mock):
+        op = DataformCreateWorkspaceOperator(
+            task_id="create-workspace",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+        )
+
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.create_workspace.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+
+class TestDataformDeleteWorkspaceOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    def test_execute(self, hook_mock):
+        op = DataformDeleteWorkspaceOperator(
+            task_id="delete-workspace",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+        )
+
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.delete_workspace.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+
+class TestDataformWriteFileOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    @mock.patch(WRITE_FILE_RESPONSE_STR)
+    def test_execute(self, _, hook_mock):
+        op = DataformWriteFileOperator(
+            task_id="write-file",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            filepath=FILEPATH,
+            contents=FILE_CONTENT,
+        )
+
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.write_file.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            filepath=FILEPATH,
+            contents=FILE_CONTENT,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+
+class TestDataformMakeDirectoryOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    @mock.patch(MAKE_DIRECTORY_RESPONSE_STR)
+    def test_execute(self, _, hook_mock):
+        op = DataformMakeDirectoryOperator(
+            task_id="write-file",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            directory_path=DIRECTORY_PATH,
+        )
+
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.make_directory.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            path=DIRECTORY_PATH,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+
+class TestDataformRemoveFileOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    def test_execute(self, hook_mock):
+        op = DataformRemoveFileOperator(
+            task_id="remove-file",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            filepath=FILEPATH,
+        )
+
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.remove_file.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            filepath=FILEPATH,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+
+class TestDataformRemoveDirectoryOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    def test_execute(self, hook_mock):
+        op = DataformRemoveDirectoryOperator(
+            task_id="remove-directory",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            directory_path=DIRECTORY_PATH,
+        )
+
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.remove_directory.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            path=DIRECTORY_PATH,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
+
+
+class TestDataformInstallNpmPackagesOperator(TestCase):
+    @mock.patch(HOOK_STR)
+    @mock.patch(INSTALL_NPM_PACKAGES_RESPONSE_STR)
+    def test_execute(self, _, hook_mock):
+        op = DataformInstallNpmPackagesOperator(
+            task_id="remove-directory",
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+        )
+
+        op.execute(context=mock.MagicMock())
+        hook_mock.return_value.install_npm_packages.assert_called_once_with(
+            project_id=PROJECT_ID,
+            region=REGION,
+            repository_id=REPOSITORY_ID,
+            workspace_id=WORKSPACE_ID,
+            retry=DEFAULT,
+            timeout=None,
+            metadata=(),
+        )
diff --git a/tests/system/providers/google/cloud/dataform/example_dataform.py b/tests/system/providers/google/cloud/dataform/example_dataform.py
index e6fcd6f4241b7..4e74644de675c 100644
--- a/tests/system/providers/google/cloud/dataform/example_dataform.py
+++ b/tests/system/providers/google/cloud/dataform/example_dataform.py
@@ -26,36 +26,72 @@
 from google.cloud.dataform_v1beta1 import WorkflowInvocation
 
 from airflow import models
-from airflow.models.baseoperator import chain
 from airflow.providers.google.cloud.operators.dataform import (
     DataformCancelWorkflowInvocationOperator,
     DataformCreateCompilationResultOperator,
+    DataformCreateRepositoryOperator,
     DataformCreateWorkflowInvocationOperator,
+    DataformCreateWorkspaceOperator,
+    DataformDeleteRepositoryOperator,
+    DataformDeleteWorkspaceOperator,
     DataformGetCompilationResultOperator,
     DataformGetWorkflowInvocationOperator,
+    DataformRemoveDirectoryOperator,
+    DataformRemoveFileOperator,
 )
 from airflow.providers.google.cloud.sensors.dataform import DataformWorkflowInvocationStateSensor
+from airflow.providers.google.cloud.utils.dataform import make_initialization_workspace_flow
+from airflow.utils.trigger_rule import TriggerRule
 
 ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
-PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT", "test-project-id")
+PROJECT_ID = os.environ.get("SYSTEM_TESTS_GCP_PROJECT", "default")
 
-DAG_ID = "dataform"
+DAG_ID = "example_dataform"
 
-REPOSITORY_ID = "dataform-test2"
+REPOSITORY_ID = f"example_dataform_repository_{ENV_ID}"
 REGION = "us-central1"
-WORKSPACE_ID = "testing"
+WORKSPACE_ID = f"example_dataform_workspace_{ENV_ID}"
 
 # This DAG is not self-run we need to do some extra configuration to execute it in automation process
-
 with models.DAG(
-    DAG_ID,
+    dag_id=DAG_ID,
+    schedule="@once",
     start_date=datetime(2021, 1, 1),
     catchup=False,
     tags=["example", "dataform"],
 ) as dag:
+    # [START howto_operator_create_repository]
+    make_repository = DataformCreateRepositoryOperator(
+        task_id="make-repository",
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+    )
+    # [END howto_operator_create_repository]
+
+    # [START howto_operator_create_workspace]
+    make_workspace = DataformCreateWorkspaceOperator(
+        task_id="make-workspace",
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+        workspace_id=WORKSPACE_ID,
+    )
+    # [END howto_operator_create_workspace]
+
+    # [START howto_initialize_workspace]
+    first_initialization_step, last_initialization_step = make_initialization_workspace_flow(
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+        workspace_id=WORKSPACE_ID,
+        package_name=f"dataform_package_{ENV_ID}",
+    )
+    # [END howto_initialize_workspace]
+
     # [START howto_operator_create_compilation_result]
     create_compilation_result = DataformCreateCompilationResultOperator(
-        task_id="create_compilation_result",
+        task_id="create-compilation-result",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
@@ -71,47 +107,47 @@
 
     # [START howto_operator_get_compilation_result]
     get_compilation_result = DataformGetCompilationResultOperator(
-        task_id="get_compilation_result",
+        task_id="get-compilation-result",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         compilation_result_id=(
-            "{{ task_instance.xcom_pull('create_compilation_result')['name'].split('/')[-1] }}"
+            "{{ task_instance.xcom_pull('create-compilation-result')['name'].split('/')[-1] }}"
         ),
     )
     # [END howto_operator_get_compilation_result]]
 
     # [START howto_operator_create_workflow_invocation]
     create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
-        task_id="create_workflow_invocation",
+        task_id='create-workflow-invocation',
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         workflow_invocation={
-            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}"
+            "compilation_result": "{{ task_instance.xcom_pull('create-compilation-result')['name'] }}"
         },
     )
     # [END howto_operator_create_workflow_invocation]
 
     # [START howto_operator_create_workflow_invocation_async]
     create_workflow_invocation_async = DataformCreateWorkflowInvocationOperator(
-        task_id="create_workflow_invocation_async",
+        task_id='create-workflow-invocation-async',
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         asynchronous=True,
         workflow_invocation={
-            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}"
+            "compilation_result": "{{ task_instance.xcom_pull('create-compilation-result')['name'] }}"
         },
     )
 
     is_workflow_invocation_done = DataformWorkflowInvocationStateSensor(
-        task_id="is_workflow_invocation_done",
+        task_id="is-workflow-invocation-done",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         workflow_invocation_id=(
-            "{{ task_instance.xcom_pull('create_workflow_invocation')['name'].split('/')[-1] }}"
+            "{{ task_instance.xcom_pull('create-workflow-invocation')['name'].split('/')[-1] }}"
         ),
         expected_statuses={WorkflowInvocation.State.SUCCEEDED},
     )
@@ -119,44 +155,99 @@
 
     # [START howto_operator_get_workflow_invocation]
     get_workflow_invocation = DataformGetWorkflowInvocationOperator(
-        task_id="get_workflow_invocation",
+        task_id='get-workflow-invocation',
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         workflow_invocation_id=(
-            "{{ task_instance.xcom_pull('create_workflow_invocation')['name'].split('/')[-1] }}"
+            "{{ task_instance.xcom_pull('create-workflow-invocation')['name'].split('/')[-1] }}"
         ),
     )
     # [END howto_operator_get_workflow_invocation]
 
-    create_second_workflow_invocation = DataformCreateWorkflowInvocationOperator(
-        task_id="create_second_workflow_invocation",
+    create_workflow_invocation_for_cancel = DataformCreateWorkflowInvocationOperator(
+        task_id='create-workflow-invocation-for-cancel',
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         workflow_invocation={
-            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}"
+            "compilation_result": "{{ task_instance.xcom_pull('create-compilation-result')['name'] }}"
         },
+        asynchronous=True,
     )
 
     # [START howto_operator_cancel_workflow_invocation]
     cancel_workflow_invocation = DataformCancelWorkflowInvocationOperator(
-        task_id="cancel_workflow_incoation",
+        task_id='cancel-workflow-invocation',
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         workflow_invocation_id=(
-            "{{ task_instance.xcom_pull('create_second_workflow_invocation')['name'].split('/')[-1] }}"
+            "{{ task_instance.xcom_pull('create-workflow-invocation-for-cancel')['name'].split('/')[-1] }}"
         ),
     )
     # [END howto_operator_cancel_workflow_invocation]
 
-    chain(
-        create_compilation_result,
-        get_compilation_result,
-        create_workflow_invocation >> get_workflow_invocation,
-        create_workflow_invocation_async >> is_workflow_invocation_done,
-        create_second_workflow_invocation >> cancel_workflow_invocation,
+    # [START howto_operator_remove_file]
+    remove_file = DataformRemoveFileOperator(
+        task_id="remove-file",
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+        workspace_id=WORKSPACE_ID,
+        filepath="package-lock.json",
+    )
+    # [END howto_operator_remove_file]
+
+    # [START howto_operator_remove_directory]
+    remove_directory = DataformRemoveDirectoryOperator(
+        task_id="remove-directory",
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+        workspace_id=WORKSPACE_ID,
+        directory_path="definitions",
+    )
+    # [END howto_operator_remove_directory]
+
+    # [START howto_operator_delete_workspace]
+    delete_workspace = DataformDeleteWorkspaceOperator(
+        task_id='delete-workspace',
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+        workspace_id=WORKSPACE_ID,
+    )
+    # [END howto_operator_delete_workspace]
+
+    delete_workspace.trigger_rule = TriggerRule.ALL_DONE
+
+    # [START howto_operator_delete_repository]
+    delete_repository = DataformDeleteRepositoryOperator(
+        task_id='delete-repository',
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+    )
+    # [END howto_operator_delete_repository]
+
+    delete_repository.trigger_rule = TriggerRule.ALL_DONE
+
+    (make_repository >> make_workspace >> first_initialization_step)
+    (
+        last_initialization_step
+        >> create_compilation_result
+        >> get_compilation_result
+        >> create_workflow_invocation
+        >> get_workflow_invocation
+        >> create_workflow_invocation_async
+        >> is_workflow_invocation_done
+        >> create_workflow_invocation_for_cancel
+        >> cancel_workflow_invocation
+        >> remove_file
+        >> remove_directory
+        >> delete_workspace
+        >> delete_repository
     )
 
     # ### Everything below this line is not part of example ###
diff --git a/tests/utils/test_dataform.py b/tests/utils/test_dataform.py
new file mode 100644
index 0000000000000..037f44d147893
--- /dev/null
+++ b/tests/utils/test_dataform.py
@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+
+from airflow.providers.google.cloud.utils.dataform import DataformLocations, define_default_location
+
+
+@pytest.mark.parametrize(
+    "region, expected",
+    [
+        ("us-central1", DataformLocations.US),
+        ("europe-west4", DataformLocations.EUROPE),
+    ],
+)
+def test_define_default_location(region, expected):
+    actual = define_default_location(region)
+    assert actual == expected

From 90cd4fc91fe6ba116c7037dd8d1530eb8e24a136 Mon Sep 17 00:00:00 2001
From: Heorhi Parkhomenka <Heorhi_Parkhomenka@epam.com>
Date: Mon, 7 Nov 2022 17:58:07 +0100
Subject: [PATCH 2/4] Black reformat

---
 airflow/providers/google/cloud/hooks/dataform.py   |  6 +++---
 airflow/providers/google/cloud/utils/dataform.py   | 10 +++++-----
 .../providers/google/cloud/hooks/test_dataform.py  |  4 ++--
 .../google/cloud/dataform/example_dataform.py      | 14 +++++++-------
 4 files changed, 17 insertions(+), 17 deletions(-)

diff --git a/airflow/providers/google/cloud/hooks/dataform.py b/airflow/providers/google/cloud/hooks/dataform.py
index c9c4070309f8b..a199402e46845 100644
--- a/airflow/providers/google/cloud/hooks/dataform.py
+++ b/airflow/providers/google/cloud/hooks/dataform.py
@@ -320,8 +320,8 @@ def delete_repository(
         client = self.get_dataform_client()
         name = f"projects/{project_id}/locations/{region}/repositories/{repository_id}"
         request = {
-            'name': name,
-            'force': force,
+            "name": name,
+            "force": force,
         }
 
         client.delete_repository(
@@ -357,7 +357,7 @@ def create_workspace(
         client = self.get_dataform_client()
         parent = f"projects/{project_id}/locations/{region}/repositories/{repository_id}"
 
-        request = {'parent': parent, 'workspace_id': workspace_id}
+        request = {"parent": parent, "workspace_id": workspace_id}
 
         workspace = client.create_workspace(
             request=request,
diff --git a/airflow/providers/google/cloud/utils/dataform.py b/airflow/providers/google/cloud/utils/dataform.py
index 2596f37b0167d..2f81d04f430e8 100644
--- a/airflow/providers/google/cloud/utils/dataform.py
+++ b/airflow/providers/google/cloud/utils/dataform.py
@@ -61,12 +61,12 @@ def make_initialization_workspace_flow(
         provided then workspace_id will be used.
     """
     make_definitions_directory = DataformMakeDirectoryOperator(
-        task_id='make-definitions-directory',
+        task_id="make-definitions-directory",
         project_id=project_id,
         region=region,
         repository_id=repository_id,
         workspace_id=workspace_id,
-        directory_path='definitions',
+        directory_path="definitions",
     )
 
     first_view_content = b"""
@@ -89,12 +89,12 @@ def make_initialization_workspace_flow(
         SELECT 1 as test
     """
     make_first_view_file = DataformWriteFileOperator(
-        task_id='write-first-view',
+        task_id="write-first-view",
         project_id=project_id,
         region=region,
         repository_id=repository_id,
         workspace_id=workspace_id,
-        filepath='definitions/first_view.sqlx',
+        filepath="definitions/first_view.sqlx",
         contents=first_view_content,
     )
 
@@ -113,7 +113,7 @@ def make_initialization_workspace_flow(
         region=region,
         repository_id=repository_id,
         workspace_id=workspace_id,
-        filepath='definitions/second_view.sqlx',
+        filepath="definitions/second_view.sqlx",
         contents=second_view_content,
     )
 
diff --git a/tests/providers/google/cloud/hooks/test_dataform.py b/tests/providers/google/cloud/hooks/test_dataform.py
index 07bc89110f807..335740f75980c 100644
--- a/tests/providers/google/cloud/hooks/test_dataform.py
+++ b/tests/providers/google/cloud/hooks/test_dataform.py
@@ -186,8 +186,8 @@ def test_create_workspace(self, mock_client):
 
         mock_client.return_value.create_workspace.assert_called_once_with(
             request={
-                'parent': parent,
-                'workspace_id': WORKSPACE_ID,
+                "parent": parent,
+                "workspace_id": WORKSPACE_ID,
             },
             retry=DEFAULT,
             timeout=None,
diff --git a/tests/system/providers/google/cloud/dataform/example_dataform.py b/tests/system/providers/google/cloud/dataform/example_dataform.py
index 4e74644de675c..de9abd38207e2 100644
--- a/tests/system/providers/google/cloud/dataform/example_dataform.py
+++ b/tests/system/providers/google/cloud/dataform/example_dataform.py
@@ -119,7 +119,7 @@
 
     # [START howto_operator_create_workflow_invocation]
     create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
-        task_id='create-workflow-invocation',
+        task_id="create-workflow-invocation",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
@@ -131,7 +131,7 @@
 
     # [START howto_operator_create_workflow_invocation_async]
     create_workflow_invocation_async = DataformCreateWorkflowInvocationOperator(
-        task_id='create-workflow-invocation-async',
+        task_id="create-workflow-invocation-async",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
@@ -155,7 +155,7 @@
 
     # [START howto_operator_get_workflow_invocation]
     get_workflow_invocation = DataformGetWorkflowInvocationOperator(
-        task_id='get-workflow-invocation',
+        task_id="get-workflow-invocation",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
@@ -166,7 +166,7 @@
     # [END howto_operator_get_workflow_invocation]
 
     create_workflow_invocation_for_cancel = DataformCreateWorkflowInvocationOperator(
-        task_id='create-workflow-invocation-for-cancel',
+        task_id="create-workflow-invocation-for-cancel",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
@@ -178,7 +178,7 @@
 
     # [START howto_operator_cancel_workflow_invocation]
     cancel_workflow_invocation = DataformCancelWorkflowInvocationOperator(
-        task_id='cancel-workflow-invocation',
+        task_id="cancel-workflow-invocation",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
@@ -212,7 +212,7 @@
 
     # [START howto_operator_delete_workspace]
     delete_workspace = DataformDeleteWorkspaceOperator(
-        task_id='delete-workspace',
+        task_id="delete-workspace",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
@@ -224,7 +224,7 @@
 
     # [START howto_operator_delete_repository]
     delete_repository = DataformDeleteRepositoryOperator(
-        task_id='delete-repository',
+        task_id="delete-repository",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,

From 3e0b5a044b07be7d049f022f2bcb588a5575d2c9 Mon Sep 17 00:00:00 2001
From: Heorhi Parkhomenka <Heorhi_Parkhomenka@epam.com>
Date: Tue, 8 Nov 2022 12:43:19 +0100
Subject: [PATCH 3/4] Fix docs

---
 airflow/providers/google/cloud/hooks/dataform.py     |  2 +-
 airflow/providers/google/cloud/links/dataform.py     |  2 +-
 airflow/providers/google/cloud/operators/dataform.py |  2 +-
 airflow/providers/google/cloud/utils/dataform.py     | 12 +-----------
 .../operators/cloud/dataform.rst                     |  2 +-
 5 files changed, 5 insertions(+), 15 deletions(-)

diff --git a/airflow/providers/google/cloud/hooks/dataform.py b/airflow/providers/google/cloud/hooks/dataform.py
index a199402e46845..aa8506c16ed7a 100644
--- a/airflow/providers/google/cloud/hooks/dataform.py
+++ b/airflow/providers/google/cloud/hooks/dataform.py
@@ -429,7 +429,7 @@ def write_file(
         :param repository_id: Required. The ID of the Dataform repository where workspace located.
         :param workspace_id: Required. The ID of the Dataform workspace where files should be created.
         :param filepath: Required. Path to file including name of the file relative to workspace root.
-        :param contents: Required. Content of the file to be writed.
+        :param contents: Required. Content of the file to be written.
         :param retry: Designation of what errors, if any, should be retried.
         :param timeout: The timeout for this request.
         :param metadata: Strings which should be sent along with the request as metadata.
diff --git a/airflow/providers/google/cloud/links/dataform.py b/airflow/providers/google/cloud/links/dataform.py
index 53dcbed11b835..5e8e8bd658f9b 100644
--- a/airflow/providers/google/cloud/links/dataform.py
+++ b/airflow/providers/google/cloud/links/dataform.py
@@ -100,7 +100,7 @@ def persist(
 
 
 class DataformWorkspaceLink(BaseGoogleLink):
-    """Helper class for constructing Datafrom workspace link."""
+    """Helper class for constructing Dataform workspace link."""
 
     name = "Dataform Workspace"
     key = "dataform_workspace"
diff --git a/airflow/providers/google/cloud/operators/dataform.py b/airflow/providers/google/cloud/operators/dataform.py
index d6801031596bf..3c92d727dd5c7 100644
--- a/airflow/providers/google/cloud/operators/dataform.py
+++ b/airflow/providers/google/cloud/operators/dataform.py
@@ -768,7 +768,7 @@ class DataformWriteFileOperator(BaseOperator):
     :param repository_id: Required. The ID of the Dataform repository where workspace located.
     :param workspace_id: Required. The ID of the Dataform workspace where files should be created.
     :param filepath: Required. Path to file including name of the file relative to workspace root.
-    :param contents: Required. Content of the file to be writed.
+    :param contents: Required. Content of the file to be written.
     :param retry: Designation of what errors, if any, should be retried.
     :param timeout: The timeout for this request.
     :param metadata: Strings which should be sent along with the request as metadata.
diff --git a/airflow/providers/google/cloud/utils/dataform.py b/airflow/providers/google/cloud/utils/dataform.py
index 2f81d04f430e8..082f57eb701f1 100644
--- a/airflow/providers/google/cloud/utils/dataform.py
+++ b/airflow/providers/google/cloud/utils/dataform.py
@@ -44,21 +44,11 @@ def make_initialization_workspace_flow(
 ) -> tuple:
     """
     Creates flow which simulates the initialization of the default project.
-    As a result project with the layout listed below will be created.
-    - definitions/
-      |-first_view.sqlx
-      |-second_view.sqlx
-    - includes/
-    - .gitignore
-    - dataform.json
-    - package.json
-    - package-lock.json
     :param project_id: Required. The ID of the Google Cloud project where workspace located.
     :param region: Required. The ID of the Google Cloud region where workspace located.
     :param repository_id: Required. The ID of the Dataform repository where workspace located.
     :param workspace_id: Required. The ID of the Dataform workspace which requires initialization.
-    :param package_name: Name of the package. This value will be used in package.json. If value is not
-        provided then workspace_id will be used.
+    :param package_name: Name of the package. If value is not provided then workspace_id will be used.
     """
     make_definitions_directory = DataformMakeDirectoryOperator(
         task_id="make-definitions-directory",
diff --git a/docs/apache-airflow-providers-google/operators/cloud/dataform.rst b/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
index ffb7505c80243..d804113e34aeb 100644
--- a/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
+++ b/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
@@ -155,7 +155,7 @@ Deletes workspace. Example of usage can be seen below:
     :language: python
     :dedent: 0
     :start-after: [START howto_operator_delete_repository]
-    :end-before: [END howto_operator_create_workspace]
+    :end-before: [END howto_operator_delete_repository]
 
 Remove file
 -----------

From 330ef7f14166e5e2cbe35b9854d183fb54905153 Mon Sep 17 00:00:00 2001
From: Heorhi Parkhomenka <Heorhi_Parkhomenka@epam.com>
Date: Thu, 10 Nov 2022 12:58:02 +0100
Subject: [PATCH 4/4] Change system test so some operators would appear in
 example

---
 .../providers/google/cloud/utils/dataform.py  | 28 +++++----
 .../operators/cloud/dataform.rst              | 36 +++++++++++
 .../google/cloud/dataform/example_dataform.py | 59 ++++++++++++++++---
 3 files changed, 103 insertions(+), 20 deletions(-)

diff --git a/airflow/providers/google/cloud/utils/dataform.py b/airflow/providers/google/cloud/utils/dataform.py
index 082f57eb701f1..d47b8228a1a08 100644
--- a/airflow/providers/google/cloud/utils/dataform.py
+++ b/airflow/providers/google/cloud/utils/dataform.py
@@ -41,6 +41,7 @@ def make_initialization_workspace_flow(
     repository_id: str,
     workspace_id: str,
     package_name: str | None = None,
+    without_installation: bool = False,
 ) -> tuple:
     """
     Creates flow which simulates the initialization of the default project.
@@ -49,6 +50,7 @@ def make_initialization_workspace_flow(
     :param repository_id: Required. The ID of the Dataform repository where workspace located.
     :param workspace_id: Required. The ID of the Dataform workspace which requires initialization.
     :param package_name: Name of the package. If value is not provided then workspace_id will be used.
+    :param without_installation: Defines should installation of npm packages be added to flow.
     """
     make_definitions_directory = DataformMakeDirectoryOperator(
         task_id="make-definitions-directory",
@@ -62,7 +64,7 @@ def make_initialization_workspace_flow(
     first_view_content = b"""
         -- This is an example SQLX file to help you learn the basics of Dataform.
         -- Visit https://cloud.google.com/dataform/docs/how-to for more information on how to configure
-            your SQL workflow.
+        --   your SQL workflow.
 
         -- You can delete this file, then commit and push your changes to your repository when you are ready.
 
@@ -93,7 +95,7 @@ def make_initialization_workspace_flow(
 
         -- Use the ref() function to manage dependencies.
         -- Learn more about ref() and other built in functions
-            here: https://cloud.google.com/dataform/docs/dataform-core
+        --  here: https://cloud.google.com/dataform/docs/dataform-core
 
         SELECT test from ${ref("first_view")}
     """
@@ -170,14 +172,6 @@ def make_initialization_workspace_flow(
         contents=package_json_content,
     )
 
-    install_npm_packages = DataformInstallNpmPackagesOperator(
-        task_id="install-npm-packages",
-        project_id=project_id,
-        region=region,
-        repository_id=repository_id,
-        workspace_id=workspace_id,
-    )
-
     (
         make_definitions_directory
         >> make_first_view_file
@@ -185,10 +179,20 @@ def make_initialization_workspace_flow(
         >> make_gitignore_file
         >> make_dataform_config_file
         >> make_package_json_file
-        >> install_npm_packages
-        >> make_includes_directory
     )
 
+    if without_installation:
+        make_package_json_file >> make_includes_directory
+    else:
+        install_npm_packages = DataformInstallNpmPackagesOperator(
+            task_id="install-npm-packages",
+            project_id=project_id,
+            region=region,
+            repository_id=repository_id,
+            workspace_id=workspace_id,
+        )
+        make_package_json_file >> install_npm_packages >> make_includes_directory
+
     return make_definitions_directory, make_includes_directory
 
 
diff --git a/docs/apache-airflow-providers-google/operators/cloud/dataform.rst b/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
index d804113e34aeb..db8e0f737651d 100644
--- a/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
+++ b/docs/apache-airflow-providers-google/operators/cloud/dataform.rst
@@ -192,3 +192,39 @@ Creates default projects structure for provided workspace. Before it can be done
     :dedent: 0
     :start-after: [START howto_initialize_workspace]
     :end-before: [END howto_initialize_workspace]
+
+Write file to workspace
+-----------------------
+Writes file with given content to specified workspace.
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformWriteFileOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_write_file]
+    :end-before: [END howto_operator_write_file]
+
+Make directory in workspace
+---------------------------
+Make directory with given path in specified workspace.
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformMakeDirectoryOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_make_directory]
+    :end-before: [END howto_operator_make_directory]
+
+Install NPM packages
+--------------------
+Installs npm packages for specified workspace
+
+:class:`~airflow.providers.google.cloud.operators.dataform.DataformInstallNpmPackagesOperator`
+
+.. exampleinclude:: /../../tests/system/providers/google/cloud/dataform/example_dataform.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_install_npm_packages]
+    :end-before: [END howto_operator_install_npm_packages]
diff --git a/tests/system/providers/google/cloud/dataform/example_dataform.py b/tests/system/providers/google/cloud/dataform/example_dataform.py
index de9abd38207e2..18a738917a3b7 100644
--- a/tests/system/providers/google/cloud/dataform/example_dataform.py
+++ b/tests/system/providers/google/cloud/dataform/example_dataform.py
@@ -36,8 +36,11 @@
     DataformDeleteWorkspaceOperator,
     DataformGetCompilationResultOperator,
     DataformGetWorkflowInvocationOperator,
+    DataformInstallNpmPackagesOperator,
+    DataformMakeDirectoryOperator,
     DataformRemoveDirectoryOperator,
     DataformRemoveFileOperator,
+    DataformWriteFileOperator,
 )
 from airflow.providers.google.cloud.sensors.dataform import DataformWorkflowInvocationStateSensor
 from airflow.providers.google.cloud.utils.dataform import make_initialization_workspace_flow
@@ -86,9 +89,20 @@
         repository_id=REPOSITORY_ID,
         workspace_id=WORKSPACE_ID,
         package_name=f"dataform_package_{ENV_ID}",
+        without_installation=True,
     )
     # [END howto_initialize_workspace]
 
+    # [START howto_operator_install_npm_packages]
+    install_npm_packages = DataformInstallNpmPackagesOperator(
+        task_id="install-npm-packages",
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+        workspace_id=WORKSPACE_ID,
+    )
+    # [END howto_operator_install_npm_packages]
+
     # [START howto_operator_create_compilation_result]
     create_compilation_result = DataformCreateCompilationResultOperator(
         task_id="create-compilation-result",
@@ -188,25 +202,51 @@
     )
     # [END howto_operator_cancel_workflow_invocation]
 
+    # [START howto_operator_make_directory]
+    make_test_directory = DataformMakeDirectoryOperator(
+        task_id="make-test-directory",
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+        workspace_id=WORKSPACE_ID,
+        directory_path="test",
+    )
+    # [END howto_operator_make_directory]
+
+    # [START howto_operator_write_file]
+    test_file_content = b"""
+    test test for test file
+    """
+    write_test_file = DataformWriteFileOperator(
+        task_id="make-test-file",
+        project_id=PROJECT_ID,
+        region=REGION,
+        repository_id=REPOSITORY_ID,
+        workspace_id=WORKSPACE_ID,
+        filepath="test/test.txt",
+        contents=test_file_content,
+    )
+    # [END howto_operator_write_file]
+
     # [START howto_operator_remove_file]
-    remove_file = DataformRemoveFileOperator(
-        task_id="remove-file",
+    remove_test_file = DataformRemoveFileOperator(
+        task_id="remove-test-file",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         workspace_id=WORKSPACE_ID,
-        filepath="package-lock.json",
+        filepath="test/test.txt",
     )
     # [END howto_operator_remove_file]
 
     # [START howto_operator_remove_directory]
-    remove_directory = DataformRemoveDirectoryOperator(
-        task_id="remove-directory",
+    remove_test_directory = DataformRemoveDirectoryOperator(
+        task_id="remove-test-directory",
         project_id=PROJECT_ID,
         region=REGION,
         repository_id=REPOSITORY_ID,
         workspace_id=WORKSPACE_ID,
-        directory_path="definitions",
+        directory_path="test",
     )
     # [END howto_operator_remove_directory]
 
@@ -236,6 +276,7 @@
     (make_repository >> make_workspace >> first_initialization_step)
     (
         last_initialization_step
+        >> install_npm_packages
         >> create_compilation_result
         >> get_compilation_result
         >> create_workflow_invocation
@@ -244,8 +285,10 @@
         >> is_workflow_invocation_done
         >> create_workflow_invocation_for_cancel
         >> cancel_workflow_invocation
-        >> remove_file
-        >> remove_directory
+        >> make_test_directory
+        >> write_test_file
+        >> remove_test_file
+        >> remove_test_directory
         >> delete_workspace
         >> delete_repository
     )
