Repository: ansible/awx
File: awxkit/awxkit/api/pages/base.py (Python)

Code before:
        else:
            HTTPBasicAuth(username, password)(req)
            resp = self.connection.post(
                auth_urls['personal_token'],
                json={"description": description, "application": None, "scope": scope},
                headers=req.headers,
            )
        if resp.ok:
            result = resp.json()
            if client_id:
                return result.pop('access_token', None)
            else:
                return result.pop('token', None)
        else:
            raise exception_from_status_code(resp.status_code)

    def get_oauth2_token(self, username='', password='', client_id=None, description='AWX CLI', client_secret=None, scope='write'):
        default_cred = config.credentials.default
        username = username or default_cred.username
        password = password or default_cred.password
        # Try gateway first, fallback to controller
        urls: AuthUrls = {"access_token": "/o/token/", "personal_token": f"{config.gateway_base_path}v1/tokens/"}
        try:
            return self._request_token(urls, username, password, client_id, description, client_secret, scope)
        except exc.NotFound:
            urls: AuthUrls = {
                "access_token": f"{config.api_base_path}o/token/",
                "personal_token": f"{config.api_base_path}v2/users/{username}/personal_tokens/",
            }
            return self._request_token(urls, username, password, client_id, description, client_secret, scope)

    def load_session(self, username='', password=''):
        default_cred = config.credentials.default
        self.connection.login(
            username=username or default_cred.username, password=password or default_cred.password, **self.connection.get_session_requirements()
        )
        return self

    def cleanup(self):
        log.debug('{0.endpoint} cleaning up.'.format(self))
        return self._cleanup(self.delete)

    def silent_cleanup(self):
        log.debug('{0.endpoint} silently cleaning up.'.format(self))
        return self._cleanup(self.silent_delete)

    def _cleanup(self, delete_method):
        try:
            delete_method()
        except exc.Forbidden as e:
            if e.msg == {'detail': 'Cannot delete running job resource.'}:

Review comment (refactor):
Since the type is already declared in the outer scope, it shouldn't be necessary to override it here with a duplicate:
```suggestion
            urls = {
```

Also, could you cover this branch with tests?

Code after:
        else:
            HTTPBasicAuth(username, password)(req)
            resp = self.connection.post(
                auth_urls['personal_token'],
                json={"description": description, "application": None, "scope": scope},
                headers=req.headers,
            )
        if resp.ok:
            result = resp.json()
            if client_id:
                return result.pop('access_token', None)
            else:
                return result.pop('token', None)
        else:
            raise exception_from_status_code(resp.status_code)

    def get_oauth2_token(self, username='', password='', client_id=None, description='AWX CLI', client_secret=None, scope='write'):
        default_cred = config.credentials.default
        username = username or default_cred.username
        password = password or default_cred.password
        # Try gateway first, fallback to controller
        urls: AuthUrls = {"access_token": "/o/token/", "personal_token": f"{config.gateway_base_path}v1/tokens/"}
        try:
            return self._request_token(urls, username, password, client_id, description, client_secret, scope)
        except exc.NotFound:
            urls = {
                "access_token": f"{config.api_base_path}o/token/",
                "personal_token": f"{config.api_base_path}v2/users/{username}/personal_tokens/",
            }
            return self._request_token(urls, username, password, client_id, description, client_secret, scope)

    def load_session(self, username='', password=''):
        default_cred = config.credentials.default
        self.connection.login(
            username=username or default_cred.username, password=password or default_cred.password, **self.connection.get_session_requirements()
        )
        return self

    def cleanup(self):
        log.debug('{0.endpoint} cleaning up.'.format(self))
        return self._cleanup(self.delete)

    def silent_cleanup(self):
        log.debug('{0.endpoint} silently cleaning up.'.format(self))
        return self._cleanup(self.silent_delete)

    def _cleanup(self, delete_method):
        try:
            delete_method()
        except exc.Forbidden as e:
            if e.msg == {'detail': 'Cannot delete running job resource.'}: