diff --git a/ckanext/validation/plugin/__init__.py b/ckanext/validation/plugin/__init__.py
index 44b4e5ec..d648709e 100644
--- a/ckanext/validation/plugin/__init__.py
+++ b/ckanext/validation/plugin/__init__.py
@@ -117,6 +117,9 @@ def get_helpers(self):
 
     # IResourceController
 
+    resources_to_validate = {}
+    packages_to_skip = {}
+
     def _process_schema_fields(self, data_dict):
         u'''
         Normalize the different ways of providing the `schema` field
@@ -151,10 +154,11 @@ def _process_schema_fields(self, data_dict):
         return data_dict
 
     def before_create(self, context, data_dict):
-        return self._process_schema_fields(data_dict)
 
-    resources_to_validate = {}
-    packages_to_skip = {}
+        is_dataset = self._data_dict_is_dataset(data_dict)
+        if not is_dataset:
+            context["_resource_create_call"] = True
+            return self._process_schema_fields(data_dict)
 
     def after_create(self, context, data_dict):
 
@@ -267,6 +271,14 @@ def after_update(self, context, data_dict):
                 # in both cases, we don't need to validate every resource.
                 return
 
+            if context.pop("_resource_create_call", False):
+                new_resource = data_dict["resources"][-1]
+                if new_resource:
+                    # This is part of a resource_create call, we only need to validate
+                    # the new resource being created
+                    self._handle_validation_for_resource(context, new_resource)
+                    return
+
             for resource in data_dict.get(u'resources', []):
                 if resource[u'id'] in self.resources_to_validate:
                     # This is part of a resource_update call, it will be
diff --git a/ckanext/validation/tests/test_logic.py b/ckanext/validation/tests/test_logic.py
index 4b7a1ac3..90046f29 100644
--- a/ckanext/validation/tests/test_logic.py
+++ b/ckanext/validation/tests/test_logic.py
@@ -145,6 +145,60 @@ def test_resource_validation_resets_existing_validation_object(
         assert validation.report is None
         assert validation.error is None
 
+    @mock.patch("ckanext.validation.logic.enqueue_job")
+    def test_resource_validation_only_called_on_resource_created(
+        self, mock_enqueue_job
+    ):
+
+        resource1 = {"format": "CSV", "url": "https://some.url"}
+
+        dataset = factories.Dataset(resources=[resource1])
+
+        assert mock_enqueue_job.call_count == 1
+        assert mock_enqueue_job.call_args[0][1][0]["id"] == dataset["resources"][0]["id"]
+
+        mock_enqueue_job.reset_mock()
+
+        resource2 = call_action(
+            "resource_create",
+            package_id=dataset["id"],
+            name="resource_2",
+            format="CSV",
+            url="https://some.url"
+        )
+
+        assert mock_enqueue_job.call_count == 1
+        assert mock_enqueue_job.call_args[0][1][0]["id"] == resource2["id"]
+
+    @mock.patch("ckanext.validation.logic.enqueue_job")
+    def test_resource_validation_only_called_on_resource_updated(
+        self, mock_enqueue_job
+    ):
+
+        resource1 = {"name": "resource_1", "format": "CSV", "url": "https://some.url"}
+        resource2 = {"name": "resource_2", "format": "CSV", "url": "https://some.url"}
+
+        dataset = factories.Dataset(resources=[resource1, resource2])
+
+        assert mock_enqueue_job.call_count == 2
+
+        mock_enqueue_job.reset_mock()
+
+        resource_1_id = [r["id"] for r in dataset["resources"] if r["name"] == "resource_1"][0]
+
+        call_action(
+            "resource_update",
+            id=resource_1_id,
+            name="resource_1",
+            format="CSV",
+            url="https://some.updated.url",
+            description="updated"
+        )
+
+        assert mock_enqueue_job.call_count == 1
+        assert mock_enqueue_job.call_args[0][1][0]["id"] == resource_1_id
+
+
 
 @pytest.mark.usefixtures("clean_db", "validation_setup", "with_plugins")
 class TestResourceValidationShow(object):