AAP-70597 Introduce new internal classes for durable data in job preparation

awx/main/models/credential.py

@@ -243,29 +243,6 @@ def passwords_needed(self):
                 needed.append('vault_password')
         return needed
 
-    @functools.cached_property
-    def context(self):
-        """
-        Property for storing runtime context during credential resolution.
-
-        The context is a dict keyed by CredentialInputSource PK, where each value
-        is a dict of runtime fields for that input source. Example::
-
-            {
-                <input_source_pk>: {
-                    "workload_identity_token": "<jwt_token>"
-                },
-                <another_input_source_pk>: {
-                    "workload_identity_token": "<different_jwt_token>"
-                },
-            }
-
-        This structure allows each input source to have its own set of runtime
-        values, avoiding conflicts when a credential has multiple input sources
-        with different configurations (e.g., different JWT audiences).
-        """
-        return {}
-
     @cached_property
     def dynamic_input_fields(self):
         # if the credential is not yet saved we can't access the input_sources
@@ -391,7 +368,7 @@ def get_input_keys(self) -> list[str]:
     def _get_dynamic_input(self, field_name):
         for input_source in self.input_sources.all():
             if input_source.input_field_name == field_name:
-                return input_source.get_input_value(context=self.context)
+                return input_source.get_input_value()
         else:
             raise ValueError('{} is not a dynamic input field'.format(field_name))
 

awx/main/models/inventory.py

@@ -1043,32 +1043,27 @@ def cloud_credential_validation(source, cred):
 
     def get_cloud_credential(self):
         """Return the credential which is directly tied to the inventory source type."""
-        credential = None
+        injector_kind = self.injector_credential_kind()
         for cred in self.credentials.all():
-            if self.source in discover_available_cloud_provider_plugin_names():
-                if cred.kind == self.source.replace('ec2', 'aws'):
-                    credential = cred
-                    break
+            if injector_kind:
+                if cred.kind == injector_kind:
+                    return cred
             else:
                 # these need to be returned in the API credential field
                 if cred.credential_type.kind != 'vault':
-                    credential = cred
-                    break
-        return credential
+                    return cred
+        return None
+
+    def injector_credential_kind(self):
+        """Return the credential kind handled by this source's inventory injector.
 
-    def get_extra_credentials(self):
-        """Return all credentials that are not used by the inventory source injector.
-        These are all credentials that should run their own inject_credential logic.
+        Returns the kind string if this source has a dedicated injector that
+        handles its own credential injection, or None otherwise. Used by
+        TaskPrepData to exclude this credential from the generic injection loop.
         """
-        special_cred = None
         if self.source in discover_available_cloud_provider_plugin_names():
-            # these have special injection logic associated with them
-            special_cred = self.get_cloud_credential()
-        extra_creds = []
-        for cred in self.credentials.all():
-            if special_cred is None or cred.pk != special_cred.pk:
-                extra_creds.append(cred)
-        return extra_creds
+            return self.source.replace('ec2', 'aws')
+        return None
 
     @property
     def credential(self):