flynt everything

Author: filiplajszczak

pythonanywhere/api/base.py

@@ -23,7 +23,7 @@ def get_api_endpoint():
             "pythonanywhere.com"
         )
     )
-    return "https://{hostname}/api/v0/user/{{username}}/{{flavor}}/".format(hostname=hostname)
+    return f"https://{hostname}/api/v0/user/{{username}}/{{flavor}}/"
 
 
 def call_api(url, method, **kwargs):
@@ -39,16 +39,14 @@ def call_api(url, method, **kwargs):
     response = requests.request(
         method=method,
         url=url,
-        headers={"Authorization": "Token {token}".format(token=token)},
+        headers={"Authorization": f"Token {token}"},
         verify=not insecure,
         **kwargs
     )
     if response.status_code == 401:
         print(response, response.text)
         raise AuthenticationError(
-            "Authentication error {status_code} calling API: {response_text}".format(
-                status_code=response.status_code, response_text=response.text
-            )
+            f"Authentication error {response.status_code} calling API: {response.text}"
         )
     return response
 

pythonanywhere/api/schedule.py

@@ -46,8 +46,7 @@ def create(self, params):
 
         if not result.ok:
             raise Exception(
-                "POST to set new task via API failed, got {result}: "
-                "{result_text}".format(result=result, result_text=result.text)
+                f"POST to set new task via API failed, got {result}: {result.text}"
             )
 
     def get_specs(self, task_id):
@@ -57,7 +56,7 @@ def get_specs(self, task_id):
         :returns: dictionary of existing task specs"""
 
         result = call_api(
-            "{base_url}{task_id}/".format(base_url=self.base_url, task_id=task_id), "GET"
+            f"{self.base_url}{task_id}/", "GET"
         )
         if result.status_code == 200:
             return result.json()
@@ -75,16 +74,15 @@ def delete(self, task_id):
         :returns: True when API response is 204"""
 
         result = call_api(
-            "{base_url}{task_id}/".format(base_url=self.base_url, task_id=task_id), "DELETE"
+            f"{self.base_url}{task_id}/", "DELETE"
         )
 
         if result.status_code == 204:
             return True
 
         if not result.ok:
             raise Exception(
-                "DELETE via API on task {task_id} failed, got {result}: "
-                "{result_text}".format(task_id=task_id, result=result, result_text=result.text)
+                f"DELETE via API on task {task_id} failed, got {result}: {result.text}"
             )
 
     def update(self, task_id, params):
@@ -99,15 +97,13 @@ def update(self, task_id, params):
         :param params: dictionary of specs to update"""
 
         result = call_api(
-            "{base_url}{task_id}/".format(base_url=self.base_url, task_id=task_id),
+            f"{self.base_url}{task_id}/",
             "PATCH",
             json=params,
         )
         if result.status_code == 200:
             return result.json()
         else:
             raise Exception(
-                "Could not update task {task_id}. Got {result}: {content}".format(
-                    task_id=task_id, result=result, content=result.text
-                )
+                f"Could not update task {task_id}. Got {result}: {result.text}"
             )

pythonanywhere/api/webapp.py

@@ -55,9 +55,7 @@ def create(self, python_version, virtualenv_path, project_path, nuke):
         )
         if not response.ok or response.json().get("status") == "ERROR":
             raise Exception(
-                "POST to create webapp via API failed, got {response}:{response_text}".format(
-                    response=response, response_text=response.text
-                )
+                f"POST to create webapp via API failed, got {response}:{response.text}"
             )
         response = call_api(
             patch_url, "patch", data={"virtualenv_path": virtualenv_path, "source_directory": project_path}
@@ -78,7 +76,7 @@ def add_default_static_files_mappings(self, project_path):
         call_api(url, "post", json=dict(url="/media/", path=str(Path(project_path) / "media")))
 
     def reload(self):
-        print(snakesay("Reloading {domain} via API".format(domain=self.domain)))
+        print(snakesay(f"Reloading {self.domain} via API"))
         url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + self.domain + "/reload/"
         response = call_api(url, "post")
         if not response.ok:
@@ -95,13 +93,11 @@ def reload(self):
                 )
                 return
             raise Exception(
-                "POST to reload webapp via API failed, got {response}:{response_text}".format(
-                    response=response, response_text=response.text
-                )
+                f"POST to reload webapp via API failed, got {response}:{response.text}"
             )
 
     def set_ssl(self, certificate, private_key):
-        print(snakesay("Setting up SSL for {domain} via API".format(domain=self.domain)))
+        print(snakesay(f"Setting up SSL for {self.domain} via API"))
         url = get_api_endpoint().format(username=getpass.getuser(), flavor="webapps") + self.domain + "/ssl/"
         response = call_api(url, "post", json={"cert": certificate, "private_key": private_key})
         if not response.ok:
@@ -120,9 +116,7 @@ def get_ssl_info(self):
         response = call_api(url, "get")
         if not response.ok:
             raise Exception(
-                "GET SSL details via API failed, got {response}:{response_text}".format(
-                    response=response, response_text=response.text
-                )
+                f"GET SSL details via API failed, got {response}:{response.text}"
             )
 
         result = response.json()
@@ -141,43 +135,39 @@ def delete_log(self, log_type, index=0):
         else:
             print(
                 snakesay(
-                    "Deleting current {type} log file for {domain} via API".format(type=log_type, domain=self.domain)
+                    f"Deleting current {log_type} log file for {self.domain} via API"
                 )
             )
 
         if index == 1:
             url = get_api_endpoint().format(
                 username=getpass.getuser(), flavor="files"
-            ) + "path/var/log/{domain}.{type}.log.1/".format(domain=self.domain, type=log_type)
+            ) + f"path/var/log/{self.domain}.{log_type}.log.1/"
         elif index > 1:
             url = get_api_endpoint().format(
                 username=getpass.getuser(), flavor="files"
-            ) + "path/var/log/{domain}.{type}.log.{index}.gz/".format(domain=self.domain, type=log_type, index=index)
+            ) + f"path/var/log/{self.domain}.{log_type}.log.{index}.gz/"
         else:
             url = get_api_endpoint().format(
                 username=getpass.getuser(), flavor="files"
-            ) + "path/var/log/{domain}.{type}.log/".format(domain=self.domain, type=log_type)
+            ) + f"path/var/log/{self.domain}.{log_type}.log/"
         response = call_api(url, "delete")
         if not response.ok:
             raise Exception(
-                "DELETE log file via API failed, got {response}:{response_text}".format(
-                    response=response, response_text=response.text
-                )
+                f"DELETE log file via API failed, got {response}:{response.text}"
             )
 
     def get_log_info(self):
         url = get_api_endpoint().format(username=getpass.getuser(), flavor="files") + "tree/?path=/var/log/"
         response = call_api(url, "get")
         if not response.ok:
             raise Exception(
-                "GET log files info via API failed, got {response}:{response_text}".format(
-                    response=response, response_text=response.text
-                )
+                f"GET log files info via API failed, got {response}:{response.text}"
             )
         file_list = response.json()
         log_types = ["access", "error", "server"]
         logs = {"access": [], "error": [], "server": []}
-        log_prefix = "/var/log/{domain}.".format(domain=self.domain)
+        log_prefix = f"/var/log/{self.domain}."
         for file_name in file_list:
             if type(file_name) == str and file_name.startswith(log_prefix):
                 log = file_name[len(log_prefix):].split(".")

pythonanywhere/django_project.py

@@ -23,14 +23,14 @@ def create_virtualenv(self, django_version=None, nuke=False):
         elif django_version == 'latest':
             packages = 'django'
         else:
-            packages = 'django=={django_version}'.format(django_version=django_version)
+            packages = f'django=={django_version}'
         self.virtualenv.pip_install(packages)
 
 
     def detect_requirements(self):
         requirements_txt = self.project_path / 'requirements.txt'
         if requirements_txt.exists():
-            return '-r {resolved_requirements}'.format(resolved_requirements=requirements_txt.resolve())
+            return f'-r {requirements_txt.resolve()}'
         return 'django'
 
 
@@ -65,7 +65,7 @@ def update_settings_file(self):
             settings = f.read()
         new_settings = settings.replace(
             'ALLOWED_HOSTS = []',
-            'ALLOWED_HOSTS = [{domain!r}]'.format(domain=self.domain)
+            f'ALLOWED_HOSTS = [{self.domain!r}]'
         )
         new_settings += dedent(
             """
@@ -98,7 +98,7 @@ def run_migrate(self):
 
 
     def update_wsgi_file(self):
-        print(snakesay('Updating wsgi file at {wsgi_file_path}'.format(wsgi_file_path=self.wsgi_file_path)))
+        print(snakesay(f'Updating wsgi file at {self.wsgi_file_path}'))
         template = (Path(__file__).parent / 'wsgi_file_template.py').open().read()
         with self.wsgi_file_path.open('w') as f:
             f.write(template.format(project=self))

pythonanywhere/launch_bash_in_virtualenv.py

@@ -21,24 +21,24 @@ def write_temporary_bashrc(virtualenv_path, unique_id, source_directory):
     else:
         bashrc = ''
     if os.path.dirname(virtualenv_path) == os.path.join(HOME, '.virtualenvs'):
-        activate_script = u'workon {}'.format(os.path.basename(virtualenv_path))
+        activate_script = f'workon {os.path.basename(virtualenv_path)}'
     else:
         activate_script_path = os.path.join(virtualenv_path, 'bin', 'activate')
         if not os.path.exists(activate_script_path):
-            print('Could not find virtualenv activation script at {}'.format(activate_script_path))
+            print(f'Could not find virtualenv activation script at {activate_script_path}')
             sys.exit(-1)
         with open(activate_script_path) as f:
             activate_script = f.read()
 
-    tmprc = os.path.join(TMP, 'tmprc.{}'.format(unique_id))
+    tmprc = os.path.join(TMP, f'tmprc.{unique_id}')
     with open(tmprc, 'w') as f:
         f.write(bashrc)
         f.write('\n')
         f.write(activate_script)
         f.write('\n')
-        f.write('cd {}\n'.format(source_directory))
+        f.write(f'cd {source_directory}\n')
         f.write('\n')
-        f.write('rm {}'.format(tmprc))
+        f.write(f'rm {tmprc}')
     return tmprc
 
 

pythonanywhere/project.py

@@ -12,10 +12,10 @@ class Project:
     def __init__(self, domain, python_version):
         self.domain = domain
         self.python_version = python_version
-        self.project_path = Path('~/{domain}'.format(domain=domain)).expanduser()
+        self.project_path = Path(f'~/{domain}').expanduser()
         self.virtualenv = Virtualenv(self.domain, self.python_version)
         self.wsgi_file_path = Path(
-            '/var/www/{mangled_domain}_wsgi.py'.format(mangled_domain=domain.replace(".", "_"))
+            f"/var/www/{domain.replace('.', '_')}_wsgi.py"
         )
         self.webapp = Webapp(domain)
 

pythonanywhere/scripts_commons.py

@@ -45,7 +45,7 @@ class ScriptSchema(Schema):
     tabulate_format = Or(
         None,
         And(str, lambda f: f in tabulate_formats),
-        error="--format should match one of: {}".format(", ".join(tabulate_formats)),
+        error=f"--format should match one of: {', '.join(tabulate_formats)}",
     )
 
     replacements = {"--": "", "<": "", ">": ""}

pythonanywhere/snakesay.py

@@ -20,13 +20,13 @@ def speech_bubble_lines(speech):
     lines, width = rewrap(speech)
     if len(lines) <= 1:
         text = ''.join(lines)
-        yield '< {text} >'.format(text=text)
+        yield f'< {text} >'
 
     else:
         yield '  ' + '_' * width
         yield '/ ' + (' ' * width) + ' \\'
         for line in lines:
-            yield '| {} |'.format(line)
+            yield f'| {line} |'
         yield '\\ ' + (' ' * width) + ' /'
         yield '  ' + '-' * width
 

pythonanywhere/task.py

@@ -55,15 +55,13 @@ def __init__(self):
     def __repr__(self):
         enabled = "enabled" if self.enabled else "disabled"
         status = (
-            "{} at {}".format(enabled, self.printable_time)
+            f"{enabled} at {self.printable_time}"
             if self.printable_time
             else "ready to be created"
         )
-        num = " <{}>:".format(self.task_id) if self.task_id else ""
+        num = f" <{self.task_id}>:" if self.task_id else ""
 
-        return "{interval} task{num} '{command}' {status}".format(
-            interval=self.interval.title(), num=num, command=self.command, status=status
-        )
+        return f"{self.interval.title()} task{num} '{self.command}' {status}"
 
     @classmethod
     def from_id(cls, task_id):
@@ -164,7 +162,7 @@ def delete_schedule(self):
         *Note*: use this method on `Task.from_id` instance."""
 
         if self.schedule.delete(self.task_id):
-            logger.info(snakesay("Task {} deleted!".format(self.task_id)))
+            logger.info(snakesay(f"Task {self.task_id} deleted!"))
 
     def update_schedule(self, params, *, porcelain=False):
         """Updates existing task using `params`.
@@ -208,14 +206,14 @@ def update_schedule(self, params, *, porcelain=False):
         }
 
         def make_spec_str(key, old_spec, new_spec):
-            return "<{}> from '{}' to '{}'".format(key, old_spec, new_spec)
+            return f"<{key}> from '{old_spec}' to '{new_spec}'"
 
         updated = [make_spec_str(key, val[0], val[1]) for key, val in diff.items()]
 
         def make_msg(join_with):
             fill = " " if join_with == ", " else join_with
-            intro = "Task {} updated:{}".format(self.task_id, fill)
-            return "{}{}".format(intro, join_with.join(updated))
+            intro = f"Task {self.task_id} updated:{fill}"
+            return f"{intro}{join_with.join(updated)}"
 
         if updated:
             if porcelain:

pythonanywhere/utils.py

@@ -6,6 +6,6 @@ def ensure_domain(domain):
     if domain == "your-username.pythonanywhere.com":
         username = getpass.getuser().lower()
         pa_domain = os.environ.get("PYTHONANYWHERE_DOMAIN", "pythonanywhere.com")
-        return "{username}.{pa_domain}".format(username=username, pa_domain=pa_domain)
+        return f"{username}.{pa_domain}"
     else:
         return domain