googleapis
diff --git a/‎google/cloud/aiplatform/base.py
+19 b/‎google/cloud/aiplatform/base.py
+19
diff --git a/‎google/cloud/aiplatform/jobs.py
+7-3 b/‎google/cloud/aiplatform/jobs.py
+7-3
@@ -21,6 +21,7 @@
 import functools
 import inspect
 import logging
+import re
 import sys
 import threading
 import time
@@ -454,6 +455,24 @@ def _format_resource_name_method(self) -> str:
     # to use custom resource id validators per resource
     _resource_id_validator: Optional[Callable[[str], None]] = None
 
+    @staticmethod
+    def _revisioned_resource_id_validator(
+        resource_id: str,
+    ) -> None:
+        """Some revisioned resource names can have '@' in them
+        to separate the resource ID from the revision ID.
+        Thus, they need their own resource id validator.
+        See https://google.aip.dev/162
+
+        Args:
+            resource_id(str): A resource ID for a resource type that accepts revision syntax.
+                See https://google.aip.dev/162.
+        Raises:
+            ValueError: If a `resource_id` doesn't conform to appropriate revision syntax.
+        """
+        if not re.compile(r"^[\w-]+@?[\w-]+$").match(resource_id):
+            raise ValueError(f"Resource {resource_id} is not a valid resource ID.")
+
     def __init__(
         self,
         project: Optional[str] = None,
 
@@ -392,6 +392,8 @@ def create(
                 Required. A fully-qualified model resource name or model ID.
                 Example: "projects/123/locations/us-central1/models/456" or
                 "456" when project and location are initialized or passed.
+                May optionally contain a version ID or alias in
+                {model_name}@{version} form.
 
                 Or an instance of aiplatform.Model.
             instances_format (str):
@@ -564,6 +566,7 @@ def create(
                 format_resource_name_method=aiplatform.Model._format_resource_name,
                 project=project,
                 location=location,
+                resource_id_validator=super()._revisioned_resource_id_validator,
             )
 
         # Raise error if both or neither source URIs are provided
@@ -713,7 +716,9 @@ def _create(
                 Required. BatchPredictionJob without _gca_resource populated.
             model_or_model_name (Union[str, aiplatform.Model]):
                 Required. Required. A fully-qualified model resource name or
-                an instance of aiplatform.Model.
+                an instance of aiplatform.Model. If a resource name, it may
+                optionally contain a version ID or alias in
+                {model_name}@{version} form.
             gca_batch_prediction_job (gca_bp_job.BatchPredictionJob):
                 Required. a batch prediction job proto for creating a batch prediction job on Vertex AI.
             generate_explanation (bool):
@@ -731,7 +736,6 @@ def _create(
                 provided instances_format or predictions_format are not supported
                 by Vertex AI.
         """
-        # select v1beta1 if explain else use default v1
 
         parent = initializer.global_config.common_location_path(
             project=empty_batch_prediction_job.project,
@@ -741,7 +745,7 @@ def _create(
         model_resource_name = (
             model_or_model_name
             if isinstance(model_or_model_name, str)
-            else model_or_model_name.resource_name
+            else model_or_model_name.versioned_resource_name
         )
 
         gca_batch_prediction_job.model = model_resource_name