fstrings with flynt (pycontribs#1044)

Co-authored-by: Neefs <[email protected]>

jira/client.py

@@ -186,11 +186,7 @@ def __call__(self, req):
         # Per Atlassian docs, use %20 for whitespace when generating qsh for URL
         # https://developer.atlassian.com/cloud/jira/platform/understanding-jwt/#qsh
         query = "&".join(sorted(parse_result.query.split("&"))).replace("+", "%20")
-        qsh = "%(method)s&%(path)s&%(query)s" % {
-            "method": req.method.upper(),
-            "path": path,
-            "query": query,
-        }
+        qsh = f"{req.method.upper()}&{path}&{query}"
 
         return hashlib.sha256(qsh.encode("utf-8")).hexdigest()
 
@@ -477,7 +473,7 @@ def __init__(
                 auth_method = (
                     oauth or basic_auth or jwt or kerberos or auth or "anonymous"
                 )
-                raise JIRAError("Can not log in with %s" % str(auth_method))
+                raise JIRAError(f"Can not log in with {str(auth_method)}")
 
         self.deploymentType = None
         if get_server_info:
@@ -558,7 +554,7 @@ def _check_for_html_error(self, content: str):
         # embedding the error in a huge webpage.
         if "<!-- SecurityTokenMissing -->" in content:
             self.log.warning("Got SecurityTokenMissing")
-            raise JIRAError("SecurityTokenMissing: %s" % content)
+            raise JIRAError(f"SecurityTokenMissing: {content}")
             return False
         return True
 
@@ -893,7 +889,7 @@ def file_stream() -> MultipartEncoder:
 
         js: Union[Dict[str, Any], List[Dict[str, Any]]] = json_loads(r)
         if not js or not isinstance(js, Iterable):
-            raise JIRAError("Unable to parse JSON: %s" % js)
+            raise JIRAError(f"Unable to parse JSON: {js}")
         jira_attachment = Attachment(
             self._options, self._session, js[0] if isinstance(js, List) else js
         )
@@ -1131,7 +1127,7 @@ def update_filter(
         data["jql"] = jql or filter.jql
         data["favourite"] = favourite or filter.favourite
 
-        url = self._get_url("filter/%s" % filter_id)
+        url = self._get_url(f"filter/{filter_id}")
         r = self._session.put(
             url, headers={"content-type": "application/json"}, data=json.dumps(data)
         )
@@ -1206,7 +1202,7 @@ def group_members(self, group: str) -> OrderedDict:
         while end_index < size - 1:
             params = {
                 "groupname": group,
-                "expand": "users[%s:%s]" % (end_index + 1, end_index + 50),
+                "expand": f"users[{end_index + 1}:{end_index + 50}]",
             }
             r2 = self._get_json("group", params=params)
             for user in r2["users"]["items"]:
@@ -1302,7 +1298,7 @@ def create_issue(
         self,
         fields: Optional[Dict[str, Any]] = None,
         prefetch: bool = True,
-        **fieldargs
+        **fieldargs,
     ) -> Issue:
         """Create a new issue and return an issue Resource for it.
 
@@ -1892,7 +1888,7 @@ def transition_issue(
         fields: Optional[Dict[str, Any]] = None,
         comment: Optional[str] = None,
         worklog: Optional[str] = None,
-        **fieldargs
+        **fieldargs,
     ):
         """Perform a transition on an issue.
 
@@ -1916,7 +1912,7 @@ def transition_issue(
             # cannot cast to int, so try to find transitionId by name
             transitionId = self.find_transitionid_by_name(issue, transition)
             if transitionId is None:
-                raise JIRAError("Invalid transition name. %s" % transition)
+                raise JIRAError(f"Invalid transition name. {transition}")
 
         data: Dict[str, Any] = {"transition": {"id": transitionId}}
         if comment:
@@ -1936,7 +1932,7 @@ def transition_issue(
         try:
             r_json = json_loads(r)
         except ValueError as e:
-            self.log.error("%s\n%s" % (e, r.text))
+            self.log.error(f"{e}\n{r.text}")
             raise e
         return r_json
 
@@ -2105,7 +2101,7 @@ def add_worklog(
             data["updateAuthor"] = data["author"]
         # report bug to Atlassian: author and updateAuthor parameters are
         # ignored.
-        url = self._get_url("issue/{0}/worklog".format(issue))
+        url = self._get_url(f"issue/{issue}/worklog")
         r = self._session.post(url, params=params, data=json.dumps(data))
 
         return Worklog(self._options, self._session, json_loads(r))
@@ -2245,9 +2241,9 @@ def issue_type_by_name(self, name: str) -> IssueType:
         if len(matching_issue_types) == 1:
             return matching_issue_types[0]
         elif len(matching_issue_types) == 0:
-            raise KeyError("Issue type '%s' is unknown." % name)
+            raise KeyError(f"Issue type '{name}' is unknown.")
         else:
-            raise KeyError("Issue type '%s' appears more than once." % name)
+            raise KeyError(f"Issue type '{name}' appears more than once.")
 
     def request_types(self, service_desk: ServiceDesk) -> List[RequestType]:
         """Returns request types supported by a service desk instance.
@@ -2262,7 +2258,7 @@ def request_types(self, service_desk: ServiceDesk) -> List[RequestType]:
             service_desk = service_desk.id
         url = (
             self.server_url
-            + "/rest/servicedeskapi/servicedesk/%s/requesttype" % service_desk
+            + f"/rest/servicedeskapi/servicedesk/{service_desk}/requesttype"
         )
         headers = {"X-ExperimentalApi": "opt-in"}
         r_json = json_loads(self._session.get(url, headers=headers))
@@ -2277,7 +2273,7 @@ def request_type_by_name(self, service_desk: ServiceDesk, name: str):
         try:
             request_type = [rt for rt in request_types if rt.name == name][0]
         except IndexError:
-            raise KeyError("Request type '%s' is unknown." % name)
+            raise KeyError(f"Request type '{name}' is unknown.")
         return request_type
 
     # User permissions
@@ -2572,7 +2568,7 @@ def project_role(self, project: str, id: str) -> Role:
             id (str): ID of the role to get
         """
         if isinstance(id, Number):
-            id = "%s" % id
+            id = f"{id}"
         return self._find_for_resource(Role, (project, id))
 
     # Resolutions
@@ -3314,7 +3310,7 @@ def _get_json(
         try:
             r_json = json_loads(r)
         except ValueError as e:
-            self.log.error("%s\n%s" % (e, r.text if r else r))
+            self.log.error(f"{e}\n{r.text if r else r}")
             raise e
         return r_json
 
@@ -3401,7 +3397,7 @@ def rename_user(self, old_user: str, new_user: str):
             params = {"username": old_user}
 
             # raw displayName
-            self.log.debug("renaming %s" % self.user(old_user).emailAddress)
+            self.log.debug(f"renaming {self.user(old_user).emailAddress}")
 
             r = self._session.put(url, params=params, data=json.dumps(payload))
             raise_on_error(r)
@@ -3421,7 +3417,7 @@ def delete_user(self, username: str) -> bool:
 
         """
 
-        url = self._get_latest_url("user/?username=%s" % username)
+        url = self._get_latest_url(f"user/?username={username}")
 
         r = self._session.delete(url)
         if 200 <= r.status_code <= 299:
@@ -3449,9 +3445,10 @@ def deactivate_user(self, username: str) -> Union[str, int]:
                     user.raw["session"]["name"],
                     user.raw["session"]["value"],
                 )
-            url = self._options[
-                "server"
-            ] + "/admin/rest/um/1/user/deactivate?username=%s" % (username)
+            url = (
+                self._options["server"]
+                + f"/admin/rest/um/1/user/deactivate?username={username}"
+            )
             # We can't use our existing session here - this endpoint is fragile and objects to extra headers
             try:
                 r = requests.post(
@@ -3467,13 +3464,12 @@ def deactivate_user(self, username: str) -> Union[str, int]:
                     return True
                 else:
                     self.log.warning(
-                        "Got response from deactivating %s: %s"
-                        % (username, r.status_code)
+                        f"Got response from deactivating {username}: {r.status_code}"
                     )
                     return r.status_code
             except Exception as e:
-                self.log.error("Error Deactivating %s: %s" % (username, e))
-                raise JIRAError("Error Deactivating %s: %s" % (username, e))
+                self.log.error(f"Error Deactivating {username}: {e}")
+                raise JIRAError(f"Error Deactivating {username}: {e}")
         else:
             url = self.server_url + "/secure/admin/user/EditUser.jspa"
             self._options["headers"][
@@ -3496,13 +3492,12 @@ def deactivate_user(self, username: str) -> Union[str, int]:
                     return True
                 else:
                     self.log.warning(
-                        "Got response from deactivating %s: %s"
-                        % (username, r.status_code)
+                        f"Got response from deactivating {username}: {r.status_code}"
                     )
                     return r.status_code
             except Exception as e:
-                self.log.error("Error Deactivating %s: %s" % (username, e))
-                raise JIRAError("Error Deactivating %s: %s" % (username, e))
+                self.log.error(f"Error Deactivating {username}: {e}")
+                raise JIRAError(f"Error Deactivating {username}: {e}")
 
     def reindex(self, force: bool = False, background: bool = True) -> bool:
         """Start jira re-indexing. Returns True if reindexing is in progress or not needed, or False.
@@ -3567,7 +3562,7 @@ def backup(self, filename: str = "backup.zip", attachments: bool = False):
             if r.status_code == 200:
                 return True
             else:
-                self.log.warning("Got %s response from calling backup." % r.status_code)
+                self.log.warning(f"Got {r.status_code} response from calling backup.")
                 return r.status_code
         except Exception as e:
             self.log.error("I see %s", e)
@@ -3625,7 +3620,7 @@ def backup_download(self, filename: str = None):
         local_file = filename or remote_file
         url = self.server_url + "/webdav/backupmanager/" + remote_file
         try:
-            self.log.debug("Writing file to %s" % local_file)
+            self.log.debug(f"Writing file to {local_file}")
             with open(local_file, "wb") as file:
                 try:
                     resp = self._session.get(
@@ -3634,12 +3629,12 @@ def backup_download(self, filename: str = None):
                 except Exception:
                     raise JIRAError()
                 if not resp.ok:
-                    self.log.error("Something went wrong with download: %s" % resp.text)
+                    self.log.error(f"Something went wrong with download: {resp.text}")
                     raise JIRAError(resp.text)
                 for block in resp.iter_content(1024):
                     file.write(block)
         except JIRAError as je:
-            self.log.error("Unable to access remote backup file: %s" % je)
+            self.log.error(f"Unable to access remote backup file: {je}")
         except IOError as ioe:
             self.log.error(ioe)
         return None
@@ -3678,7 +3673,7 @@ def delete_project(self, pid: Union[str, Project]) -> Optional[bool]:
         if isinstance(pid, Project) and hasattr(pid, "id"):
             pid = str(pid.id)
 
-        url = self._get_url("project/%s" % pid)
+        url = self._get_url(f"project/{pid}")
         r = self._session.delete(url)
         if r.status_code == 403:
             raise JIRAError("Not enough permissions to delete project")
@@ -3757,7 +3752,7 @@ def projectcategories(self):
     @lru_cache(maxsize=None)
     def avatars(self, entity="project"):
 
-        url = self._get_url("avatar/%s/system" % entity)
+        url = self._get_url(f"avatar/{entity}/system")
 
         r = self._session.get(url)
         data: Dict[str, Any] = json_loads(r)
@@ -3802,7 +3797,7 @@ def workflows(self):
 
     def delete_screen(self, id: str):
 
-        url = self._get_url("screens/%s" % id)
+        url = self._get_url(f"screens/{id}")
 
         r = self._session.delete(url)
         data = json_loads(r)
@@ -3812,7 +3807,7 @@ def delete_screen(self, id: str):
 
     def delete_permissionscheme(self, id: str):
 
-        url = self._get_url("permissionscheme/%s" % id)
+        url = self._get_url(f"permissionscheme/{id}")
 
         r = self._session.delete(url)
         data = json_loads(r)
@@ -4103,7 +4098,7 @@ def role(self) -> List[Dict[str, Any]]:
     def get_igrid(self, issueid: str, customfield: str, schemeid: str):
         url = self.server_url + "/rest/idalko-igrid/1.0/datagrid/data"
         if str(customfield).isdigit():
-            customfield = "customfield_%s" % customfield
+            customfield = f"customfield_{customfield}"
         params = {
             "_issueId": issueid,
             "_fieldId": customfield,
@@ -4246,7 +4241,7 @@ def sprints(
             return self._fetch_pages(
                 Sprint,
                 "values",
-                "board/%s/sprint" % board_id,
+                f"board/{board_id}/sprint",
                 startAt,
                 maxResults,
                 params,
@@ -4280,25 +4275,23 @@ def update_sprint(self, id, name=None, startDate=None, endDate=None, state=None)
                 )
             payload["state"] = state
 
-        url = self._get_url("sprint/%s" % id, base=self.AGILE_BASE_URL)
+        url = self._get_url(f"sprint/{id}", base=self.AGILE_BASE_URL)
         r = self._session.put(url, data=json.dumps(payload))
 
         return json_loads(r)
 
     def incompletedIssuesEstimateSum(self, board_id: str, sprint_id: str):
         """Return the total incompleted points this sprint."""
         data: Dict[str, Any] = self._get_json(
-            "rapid/charts/sprintreport?rapidViewId=%s&sprintId=%s"
-            % (board_id, sprint_id),
+            f"rapid/charts/sprintreport?rapidViewId={board_id}&sprintId={sprint_id}",
             base=self.AGILE_BASE_URL,
         )
         return data["contents"]["incompletedIssuesEstimateSum"]["value"]
 
     def removed_issues(self, board_id: str, sprint_id: str):
         """Return the completed issues for the sprint."""
         r_json: Dict[str, Any] = self._get_json(
-            "rapid/charts/sprintreport?rapidViewId=%s&sprintId=%s"
-            % (board_id, sprint_id),
+            f"rapid/charts/sprintreport?rapidViewId={board_id}&sprintId={sprint_id}",
             base=self.AGILE_BASE_URL,
         )
         issues = [
@@ -4311,8 +4304,7 @@ def removed_issues(self, board_id: str, sprint_id: str):
     def removedIssuesEstimateSum(self, board_id: str, sprint_id: str):
         """Return the total incompleted points this sprint."""
         data: Dict[str, Any] = self._get_json(
-            "rapid/charts/sprintreport?rapidViewId=%s&sprintId=%s"
-            % (board_id, sprint_id),
+            f"rapid/charts/sprintreport?rapidViewId={board_id}&sprintId={sprint_id}",
             base=self.AGILE_BASE_URL,
         )
         return data["contents"]["puntedIssuesEstimateSum"]["value"]
@@ -4428,7 +4420,7 @@ def create_sprint(
             self._options["agile_rest_path"]
             == GreenHopperResource.GREENHOPPER_REST_PATH
         ):
-            url = self._get_url("sprint/%s" % board_id, base=self.AGILE_BASE_URL)
+            url = self._get_url(f"sprint/{board_id}", base=self.AGILE_BASE_URL)
             r = self._session.post(url)
             raw_issue_json = json_loads(r)
             """ now r contains something like:
@@ -4444,7 +4436,7 @@ def create_sprint(
             }"""
 
             url = self._get_url(
-                "sprint/%s" % raw_issue_json["id"], base=self.AGILE_BASE_URL
+                f"sprint/{raw_issue_json['id']}", base=self.AGILE_BASE_URL
             )
             r = self._session.put(url, data=json.dumps(payload))
             raw_issue_json = json_loads(r)
@@ -4477,7 +4469,7 @@ def add_issues_to_sprint(self, sprint_id: int, issue_keys: List[str]) -> Respons
             Response
         """
         if self._options["agile_rest_path"] == GreenHopperResource.AGILE_BASE_REST_PATH:
-            url = self._get_url("sprint/%s/issue" % sprint_id, base=self.AGILE_BASE_URL)
+            url = self._get_url(f"sprint/{sprint_id}/issue", base=self.AGILE_BASE_URL)
             payload = {"issues": issue_keys}
             try:
                 return self._session.post(url, data=json.dumps(payload))
@@ -4535,7 +4527,7 @@ def add_issues_to_epic(
         data: Dict[str, Any] = {}
         data["issueKeys"] = issue_keys
         data["ignoreEpics"] = ignore_epics
-        url = self._get_url("epics/%s/add" % epic_id, base=self.AGILE_BASE_URL)
+        url = self._get_url(f"epics/{epic_id}/add", base=self.AGILE_BASE_URL)
         return self._session.put(url, data=json.dumps(data))
 
     # TODO(ssbarnea): Both GreenHopper and new Jira Agile API support moving more than one issue.

jira/config.py

@@ -87,7 +87,7 @@ def findfile(path):
 
     config_file = findfile("config.ini")
     if config_file:
-        logging.debug("Found %s config file" % config_file)
+        logging.debug(f"Found {config_file} config file")
 
     if not profile:
         if config_file: