Repository: ansible/awx
File: awx/main/models/credential.py (Python)

Code before:
        Create a CredentialType for discovered credential plugins.

        By default, this function will attempt to acquire the globally distributed lock (postgres advisory lock).
        If the lock is acquired the method will call the underlying method.
        If the lock is NOT acquired the method will NOT call the underlying method.

        lock=False will set acquired to True and appear to acquire the lock.
        lock=True, wait_for_lock=False will attempt to acquire the lock and NOT block.
        lock=True, wait_for_lock=True will attempt to acquire the lock and will block until the exclusive lock is acquired.

        :param lock(optional[bool]):          Attempt to acquire the postgres advisory lock.
        :param wait_for_lock(optional[bool]): Block and wait forever for the postgres advisory lock.
        """
        if apps and not apps.ready:
            return

        with advisory_lock(cls.CREDENTIAL_REGISTRATION_ADVISORY_LOCK_NAME, wait=wait_for_lock) if lock else nullcontext(True) as acquired:
            if acquired:
                cls._setup_tower_managed_defaults(apps=apps, app_config=app_config)

    @classmethod
    def load_plugin(cls, ns, plugin):
        # TODO: User "side-loaded" credential custom_injectors isn't supported
        ManagedCredentialType.registry[ns] = SimpleNamespace(namespace=ns, name=plugin.name, kind='external', inputs=plugin.inputs, backend=plugin.backend)

    def inject_credential(self, credential, env, safe_env, args, private_data_dir, **kwargs):
        from awx_plugins.interfaces._temporary_private_inject_api import inject_credential

        inject_credential(self, credential, env, safe_env, args, private_data_dir, **kwargs)


class CredentialTypeHelper:
    @classmethod
    def get_creation_params(cls, cred_type):
        if cred_type.kind == 'external':
            return dict(namespace=cred_type.namespace, kind=cred_type.kind, name=cred_type.name, managed=True)
        return dict(
            namespace=cred_type.namespace,
            kind=cred_type.kind,
            name=cred_type.name,
            managed=True,
            inputs=cred_type.inputs,
            injectors=cred_type.injectors,
        )

    @classmethod
    def create(cls, cred_type):
        res = CredentialType(**CredentialTypeHelper.get_creation_params(cred_type))
        res.custom_injectors = getattr(cred_type, "custom_injectors", None)
        return res


Review comment (bug):
Similarly to the interfaces repo, try to avoid untyped kwargs. Eventually, this will let type checkers find bugs for you.

Code after:
        Create a CredentialType for discovered credential plugins.

        By default, this function will attempt to acquire the globally distributed lock (postgres advisory lock).
        If the lock is acquired the method will call the underlying method.
        If the lock is NOT acquired the method will NOT call the underlying method.

        lock=False will set acquired to True and appear to acquire the lock.
        lock=True, wait_for_lock=False will attempt to acquire the lock and NOT block.
        lock=True, wait_for_lock=True will attempt to acquire the lock and will block until the exclusive lock is acquired.

        :param lock(optional[bool]):          Attempt to acquire the postgres advisory lock.
        :param wait_for_lock(optional[bool]): Block and wait forever for the postgres advisory lock.
        """
        if apps and not apps.ready:
            return

        with advisory_lock(cls.CREDENTIAL_REGISTRATION_ADVISORY_LOCK_NAME, wait=wait_for_lock) if lock else nullcontext(True) as acquired:
            if acquired:
                cls._setup_tower_managed_defaults(apps=apps, app_config=app_config)

    @classmethod
    def load_plugin(cls, ns, plugin):
        # TODO: User "side-loaded" credential custom_injectors isn't supported
        ManagedCredentialType.registry[ns] = SimpleNamespace(namespace=ns, name=plugin.name, kind='external', inputs=plugin.inputs, backend=plugin.backend)

    def inject_credential(self, credential, env, safe_env, args, private_data_dir, container_root=None):
        from awx_plugins.interfaces._temporary_private_inject_api import inject_credential

        inject_credential(self, credential, env, safe_env, args, private_data_dir, container_root=container_root)


class CredentialTypeHelper:
    @classmethod
    def get_creation_params(cls, cred_type):
        if cred_type.kind == 'external':
            return dict(namespace=cred_type.namespace, kind=cred_type.kind, name=cred_type.name, managed=True)
        return dict(
            namespace=cred_type.namespace,
            kind=cred_type.kind,
            name=cred_type.name,
            managed=True,
            inputs=cred_type.inputs,
            injectors=cred_type.injectors,
        )

    @classmethod
    def create(cls, cred_type):
        res = CredentialType(**CredentialTypeHelper.get_creation_params(cred_type))
        res.custom_injectors = getattr(cred_type, "custom_injectors", None)
        return res