feat(cloud): Add workspace create, delete, and rename support

airbyte/_util/api_util.py

@@ -29,6 +29,7 @@
     AirbyteError,
     AirbyteMissingResourceError,
     AirbyteMultipleResourcesError,
+    AirbyteWorkspaceNotEmptyError,
     PyAirbyteInputError,
 )
 from airbyte.secrets.base import SecretString
@@ -284,6 +285,193 @@ def get_workspace(
     )
 
 
+def create_workspace(
+    *,
+    name: str,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+    organization_id: str | None = None,
+    region_id: str | None = None,
+) -> models.WorkspaceResponse:
+    """Create a workspace."""
+    airbyte_instance = get_airbyte_server_instance(
+        api_root=api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+    )
+    base_context = {
+        "name": name,
+        "organization_id": organization_id,
+        "region_id": region_id,
+        "api_root": api_root,
+    }
+    try:
+        response = airbyte_instance.workspaces.create_workspace(
+            request=models.WorkspaceCreateRequest(
+                name=name,
+                organization_id=organization_id,
+                region_id=region_id,
+            )
+        )
+    except SDKError as e:
+        raise _wrap_sdk_error(e, base_context) from e
+
+    if status_ok(response.status_code) and response.workspace_response:
+        return response.workspace_response
+
+    raise AirbyteError(
+        message="Could not create workspace.",
+        context={
+            **base_context,
+            "request_url": response.raw_response.url,
+            "status_code": response.status_code,
+        },
+        response=response,
+    )
+
+
+def rename_workspace(
+    workspace_id: str,
+    *,
+    name: str,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+) -> models.WorkspaceResponse:
+    """Rename a workspace."""
+    airbyte_instance = get_airbyte_server_instance(
+        api_root=api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+    )
+    base_context = {
+        "workspace_id": workspace_id,
+        "name": name,
+        "api_root": api_root,
+    }
+    try:
+        response = airbyte_instance.workspaces.update_workspace(
+            api.UpdateWorkspaceRequest(
+                workspace_id=workspace_id,
+                workspace_update_request=models.WorkspaceUpdateRequest(name=name),
+            )
+        )
+    except SDKError as e:
+        raise _wrap_sdk_error(e, base_context) from e
+
+    if status_ok(response.status_code) and response.workspace_response:
+        return response.workspace_response
+
+    raise AirbyteError(
+        message="Could not rename workspace.",
+        context={
+            **base_context,
+            "request_url": response.raw_response.url,
+            "status_code": response.status_code,
+        },
+        response=response,
+    )
+
+
+def permanently_delete_workspace(
+    workspace_id: str,
+    *,
+    workspace_name: str | None = None,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+    safe_mode: bool = True,
+) -> None:
+    """Delete an empty workspace.
+
+    Args:
+        workspace_id: The workspace ID to delete.
+        workspace_name: Optional workspace name. If not provided and safe mode is enabled,
+            the workspace name is fetched from the API for safety checks.
+        api_root: The API root URL.
+        client_id: OAuth client ID.
+        client_secret: OAuth client secret.
+        bearer_token: Bearer token for authentication.
+        safe_mode: If True, the workspace name must contain `delete-me` or `deleteme`
+            (case insensitive). Defaults to True.
+
+    Raises:
+        PyAirbyteInputError: If safe mode is True and the workspace name does not meet
+            the safety requirements.
+        AirbyteWorkspaceNotEmptyError: If the workspace contains connections.
+    """
+    if safe_mode:
+        if workspace_name is None:
+            workspace_info = get_workspace(
+                workspace_id=workspace_id,
+                api_root=api_root,
+                client_id=client_id,
+                client_secret=client_secret,
+                bearer_token=bearer_token,
+            )
+            workspace_name = workspace_info.name
+
+        if not _is_safe_name_to_delete(workspace_name):
+            raise PyAirbyteInputError(
+                message=(
+                    "Cannot delete workspace with safe_mode enabled because the workspace "
+                    "name does not contain 'delete-me' or 'deleteme'."
+                ),
+                context={
+                    "workspace_id": workspace_id,
+                    "workspace_name": workspace_name,
+                    "safe_mode": True,
+                },
+            )
+
+    connections = list_connections(
+        workspace_id=workspace_id,
+        api_root=api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+        limit=1,
+    )
+    if connections:
+        raise AirbyteWorkspaceNotEmptyError(
+            workspace_id=workspace_id,
+            connection_ids=[connection.connection_id for connection in connections],
+        )
+
+    airbyte_instance = get_airbyte_server_instance(
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+        api_root=api_root,
+    )
+    base_context = {
+        "workspace_id": workspace_id,
+        "api_root": api_root,
+    }
+    try:
+        response = airbyte_instance.workspaces.delete_workspace(
+            api.DeleteWorkspaceRequest(workspace_id=workspace_id),
+        )
+    except SDKError as e:
+        raise _wrap_sdk_error(e, base_context) from e
+
+    if not status_ok(response.status_code):
+        raise AirbyteError(
+            context={
+                **base_context,
+                "request_url": response.raw_response.url,
+                "status_code": response.status_code,
+            },
+            response=response,
+        )
+
+
 # List resources
 
 

airbyte/_util/telemetry.py

@@ -103,7 +103,7 @@ def _setup_analytics() -> str | bool:
     if not _ANALYTICS_FILE.exists():
         # This is a one-time message to inform the user that we are tracking anonymous usage stats.
         print(
-            "Thank you for using PyAirbyte!\n"
+            "Thank you for using Airbyte!\n"
             "Anonymous usage reporting is currently enabled. For more information, please"
             " see https://docs.airbyte.com/telemetry",
             file=sys.stderr,

airbyte/cloud/client.py

@@ -140,6 +140,64 @@ def get_workspace(self, workspace_id: str | None = None) -> CloudWorkspace:
             config_api_root=credentials.config_api_root,
         )
 
+    def create_workspace(
+        self,
+        *,
+        name: str,
+        organization_id: str | None = None,
+        region_id: str | None = None,
+    ) -> api_imports.WorkspaceResponse:
+        """Create an Airbyte workspace."""
+        resolved_organization_id = organization_id or self.organization_id
+        return api_util.create_workspace(
+            name=name,
+            organization_id=resolved_organization_id,
+            region_id=region_id,
+            api_root=self.public_api_root,
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            bearer_token=self.bearer_token,
+        )
+
+    def rename_workspace(
+        self,
+        workspace_id: str,
+        *,
+        name: str,
+    ) -> api_imports.WorkspaceResponse:
+        """Rename an Airbyte workspace."""
+        return api_util.rename_workspace(
+            workspace_id=workspace_id,
+            name=name,
+            api_root=self.public_api_root,
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            bearer_token=self.bearer_token,
+        )
+
+    def permanently_delete_workspace(
+        self,
+        workspace_id: str,
+        *,
+        workspace_name: str | None = None,
+        safe_mode: bool = True,
+    ) -> None:
+        """Permanently delete an Airbyte workspace if it has no connections.
+
+        When `safe_mode` is enabled, the workspace name must contain `delete-me`
+        or `deleteme`. This also checks for existing connections before deleting
+        and raises `AirbyteWorkspaceNotEmptyError` if the workspace is not empty.
+        """
+        api_util.permanently_delete_workspace(
+            workspace_id=workspace_id,
+            workspace_name=workspace_name,
+            api_root=self.public_api_root,
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            bearer_token=self.bearer_token,
+            safe_mode=safe_mode,
+        )
+
     @overload
     def list_workspaces(
         self,

airbyte/cloud/workspaces.py

@@ -660,6 +660,43 @@ def list_workspaces(
             limit=limit,
         )
 
+    def rename(
+        self,
+        name: str,
+    ) -> CloudWorkspace:
+        """Rename this workspace."""
+        api_util.rename_workspace(
+            workspace_id=self.workspace_id,
+            name=name,
+            api_root=self.api_root,
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            bearer_token=self.bearer_token,
+        )
+        return self
+
+    def permanently_delete(
+        self,
+        *,
+        workspace_name: str | None = None,
+        safe_mode: bool = True,
+    ) -> None:
+        """Permanently delete this workspace if it has no connections.
+
+        When `safe_mode` is enabled, the workspace name must contain `delete-me`
+        or `deleteme`. This also checks for existing connections before deleting
+        and raises `AirbyteWorkspaceNotEmptyError` if the workspace is not empty.
+        """
+        api_util.permanently_delete_workspace(
+            workspace_id=self.workspace_id,
+            workspace_name=workspace_name,
+            api_root=self.api_root,
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            bearer_token=self.bearer_token,
+            safe_mode=safe_mode,
+        )
+
     def list_connections(
         self,
         name: str | None = None,

airbyte/exceptions.py

@@ -516,6 +516,17 @@ class AirbyteWorkspaceMismatchError(AirbyteError):
     """The workspace ID returned by the API."""
 
 
+@dataclass
+class AirbyteWorkspaceNotEmptyError(AirbyteError):
+    """Workspace cannot be deleted because it contains connections."""
+
+    workspace_id: str | None = None
+    """The workspace ID that was expected to be empty."""
+
+    connection_ids: list[str] | None = None
+    """The connection IDs found in the workspace."""
+
+
 @dataclass
 class AirbyteConnectionSyncTimeoutError(AirbyteConnectionSyncError):
     """An timeout occurred while waiting for the remote Airbyte job to complete."""