Rename the module so we can import/register it; fix command-line form…

…atting

prefect_gcp/__init__.py

@@ -1,15 +1,16 @@
-from . import _version
+from prefect._internal.compatibility.deprecated import (
+    register_renamed_module,
+)
 
-from .bigquery import BigQueryWarehouse  # noqa
+from . import _version
 from .aiplatform import VertexAICustomTrainingJob  # noqa
-from .cloud_storage import GcsBucket  # noqa
+from .bigquery import BigQueryWarehouse  # noqa
 from .cloud_run import CloudRunJob  # noqa
-from .secret_manager import GcpSecret  # noqa
+from .cloud_storage import GcsBucket  # noqa
 from .credentials import GcpCredentials  # noqa
+from .secret_manager import GcpSecret  # noqa
+from .vertex_worker import VertexAIWorker  # noqa
 from .worker import CloudRunWorker  # noqa
-from prefect._internal.compatibility.deprecated import (
-    register_renamed_module,
-)
 
 register_renamed_module(
     "prefect_gcp.projects", "prefect_gcp.deployments", start_date="Jun 2023"

prefect_gcp/vertex-worker.py → prefect_gcp/vertex_worker.py

@@ -1,7 +1,8 @@
+import shlex
 import datetime
 import re
 import time
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, MutableSequence
 from uuid import uuid4
 
 import anyio
@@ -299,6 +300,13 @@ def job_name(self) -> str:
         job_name = f"{repo_name}-{unique_suffix}"
         return job_name
 
+    def command_as_list(self) -> MutableSequence[str]:
+        if not self.command:
+            return ["python", "-m", "prefect.engine"]
+
+        return shlex.split(self.command)
+
+
 
 class VertexAIWorkerResult(BaseWorkerResult):
     """Contains information about the final state of a completed process"""
@@ -397,9 +405,10 @@ def _build_job_spec(
                 **configuration.env,
             }.items()
         ]
+
         container_spec = ContainerSpec(
             image_uri=configuration.image,
-            command=configuration.command,
+            command=configuration.command_as_list(),
             args=[],
             env=env_list,
         )