[Issue #6843] Replace concatenated strings, %-strings, and str.format…

… with f-strings (#7096)

* [Fixes #6843] Replace concatenated strings, %-strings, and str.format with f-strings

* [Fixes #6843] Replace concatenated strings, %-strings, and str.format with f-strings

* - Flake8 issues

* [Fixes #6843] Replace concatenated strings, %-strings, and str.format with f-strings

* [Fixes #6843] Replace concatenated strings, %-strings, and str.format with f-strings

.env

@@ -25,6 +25,8 @@ GEONODE_GEODATABASE=geonode_data
 GEONODE_GEODATABASE_PASSWORD=geonode_data
 GEONODE_DATABASE_SCHEMA=public
 GEONODE_GEODATABASE_SCHEMA=public
+DATABASE_HOST=db
+DATABASE_PORT=5432
 DATABASE_URL=postgis://geonode:geonode@db:5432/geonode
 GEODATABASE_URL=postgis://geonode_data:geonode_data@db:5432/geonode_data
 GEONODE_DB_CONN_MAX_AGE=0

geonode/api/resourcebase_api.py

@@ -340,12 +340,11 @@ def build_haystack_filters(self, parameters):
 
             if len(subtypes) > 0:
                 types.append("layer")
-                sqs = SearchQuerySet().narrow("subtype:%s" %
-                                              ','.join(map(str, subtypes)))
+                sqs = SearchQuerySet().narrow(f"subtype:{','.join(map(str, subtypes))}")
 
             if len(types) > 0:
                 sqs = (SearchQuerySet() if sqs is None else sqs).narrow(
-                    "type:%s" % ','.join(map(str, types)))
+                    f"type:{','.join(map(str, types))}")
 
         # Filter by Query Params
         # haystack bug? if boosted fields aren't included in the
@@ -417,7 +416,7 @@ def build_haystack_filters(self, parameters):
         if owner:
             sqs = (
                 SearchQuerySet() if sqs is None else sqs).narrow(
-                    "owner__username:%s" % ','.join(map(str, owner)))
+                    f"owner__username:{','.join(map(str, owner))}")
 
         # filter by date
         if date_start:

geonode/api/tests.py

@@ -209,7 +209,7 @@ def setUp(self):
         self.passwd = 'admin'
         self._user = get_user_model().objects.get(username=self.user)
         self.token = get_or_create_token(self._user)
-        self.auth_header = 'Bearer {}'.format(self.token)
+        self.auth_header = f'Bearer {self.token}'
         self.list_url = reverse(
             'api_dispatch_list',
             kwargs={
@@ -402,21 +402,21 @@ def to_date(val):
             return val.date().strftime(fstring)
 
         d1 = to_date(now - step)
-        filter_url = self.list_url + '?date__exact={}'.format(d1)
+        filter_url = self.list_url + f'?date__exact={d1}'
 
         resp = self.api_client.get(filter_url)
         self.assertValidJSONResponse(resp)
         self.assertEqual(len(self.deserialize(resp)['objects']), 0)
 
         d3 = to_date(now - (3 * step))
-        filter_url = self.list_url + '?date__gte={}'.format(d3)
+        filter_url = self.list_url + f'?date__gte={d3}'
 
         resp = self.api_client.get(filter_url)
         self.assertValidJSONResponse(resp)
         self.assertEqual(len(self.deserialize(resp)['objects']), 0)
 
         d4 = to_date(now - (4 * step))
-        filter_url = self.list_url + '?date__range={},{}'.format(d4, to_date(now))
+        filter_url = self.list_url + f'?date__range={d4},{to_date(now)}'
 
         resp = self.api_client.get(filter_url)
         self.assertValidJSONResponse(resp)

geonode/api/views.py

@@ -141,7 +141,7 @@ def verify_token(request):
                 token_info,
                 content_type="application/json"
             )
-            response["Authorization"] = ("Bearer %s" % access_token)
+            response["Authorization"] = f"Bearer {access_token}"
             return response
         else:
             return HttpResponse(

geonode/base/forms.py

@@ -503,7 +503,7 @@ def clean_keywords(self):
                     _kk = _kk.replace('%u', r'\u').encode('unicode-escape').replace(
                         b'\\\\u',
                         b'\\u').decode('unicode-escape') if '%u' in _kk else _kk
-                    _hk = HierarchicalKeyword.objects.filter(name__iexact='%s' % _kk.strip())
+                    _hk = HierarchicalKeyword.objects.filter(name__iexact=f'{_kk.strip()}')
                     if _hk and len(_hk) > 0:
                         _unsescaped_kwds.append(str(_hk[0]))
                     else:

geonode/base/management/commands/delete_resources.py

@@ -22,7 +22,6 @@
 import json
 
 # import needed to resolve model filters:
-from django.db.models import Q
 from django.db import transaction, IntegrityError
 from django.core.management.base import BaseCommand, CommandError
 
@@ -76,10 +75,12 @@ def handle(self, **options):
 
         # check argument set
         if all(config is None for config in {layer_filters, map_filters, document_filters, config_path}):
-            raise CommandError("No configuration provided. Please specify any of the following arguments: '-l', '-m', '-d', '-c'.")
+            raise CommandError(
+                "No configuration provided. Please specify any of the following arguments: '-l', '-m', '-d', '-c'.")
 
         if any([layer_filters, map_filters, document_filters]) and config_path:
-            raise CommandError("Too many configuration options provided. Please use either '-c' or '-l', '-m', '-d' arguments.")
+            raise CommandError(
+                "Too many configuration options provided. Please use either '-c' or '-l', '-m', '-d' arguments.")
 
         # check config_file, if it exists
         if config_path:
@@ -102,7 +103,12 @@ def handle(self, **options):
                         # or 'filters' is not set in config OR it is not an JSON object
                         or not isinstance(config.get('filters'), dict)
                         # or all filters are empty
-                        or not any([config.get('filters').get('layer'), config.get('filters').get('map'), config.get('filters').get('document')])
+                        or not any(
+                            [
+                                config.get('filters').get('layer'),
+                                config.get('filters').get('map'),
+                                config.get('filters').get('document')
+                            ])
                 ):
                     print('Nothing to be done... exiting delete_resources command.')
                     return

geonode/base/management/commands/fixoauthuri.py

@@ -53,16 +53,12 @@ def handle(self, *args, **options):
         force_exec = options.get('force_exec')
         target_address = options.get('target_address')
 
-        from django.conf import settings
         client_id = None
         client_secret = None
 
         if check_ogc_backend(geoserver.BACKEND_PACKAGE):
             from geonode.geoserver.helpers import ogc_server_settings
-            redirect_uris = '%s\n%s\n%s' % (
-                ogc_server_settings.LOCATION,
-                ogc_server_settings.public_url,
-                "{0}/geoserver/".format(target_address))
+            redirect_uris = f'{ogc_server_settings.LOCATION}\n{ogc_server_settings.public_url}\n{target_address}/geoserver/'  # noqa
             if Application.objects.filter(name='GeoServer').exists():
                 Application.objects.filter(name='GeoServer').update(redirect_uris=redirect_uris)
                 if force_exec:
@@ -86,4 +82,4 @@ def handle(self, *args, **options):
                     client_secret=client_secret,
                     user=get_user_model().objects.filter(is_superuser=True)[0]
                 )
-        return '%s,%s' % (client_id, client_secret)
+        return f'{client_id},{client_secret}'

geonode/base/management/commands/helpers.py

@@ -46,9 +46,9 @@ def confirm(prompt=None, resp=False):
         prompt = 'Confirm'
 
     if resp:
-        prompt = '%s [%s]|%s: ' % (prompt, 'y', 'n')
+        prompt = f'{prompt} [y]|n: '
     else:
-        prompt = '%s [%s]|%s: ' % (prompt, 'n', 'y')
+        prompt = f'{prompt} [n]|y: '
 
     while True:
         ans = input(prompt)

geonode/base/management/commands/load_thesaurus.py

@@ -93,7 +93,7 @@ def load_thesaurus(self, input_file, name, store):
             raise CommandError("ConceptScheme not found in file")
 
         titles = scheme.findall('dc:title', ns)
-        
+
         default_lang = getattr(settings, 'THESAURUS_DEFAULT_LANG', None)
         available_lang = get_all_lang_available_with_title(titles, LANG_ATTRIB)
         thesaurus_title = determinate_value(available_lang, default_lang)
@@ -102,7 +102,7 @@ def load_thesaurus(self, input_file, name, store):
         date_issued = scheme.find('dcterms:issued', ns).text
         about = scheme.attrib.get(ABOUT_ATTRIB)
 
-        print('Thesaurus "{}" issued on {}'.format(thesaurus_title, date_issued))
+        print(f'Thesaurus "{thesaurus_title}" issued at {date_issued}')
 
         thesaurus = Thesaurus()
         thesaurus.identifier = name
@@ -133,7 +133,7 @@ def load_thesaurus(self, input_file, name, store):
                 available_lang = get_all_lang_available_with_title(concepts, LANG_ATTRIB)
                 alt_label = determinate_value(available_lang, default_lang)
 
-            print('Concept {} ({})'.format(alt_label, about))
+            print(f'Concept {alt_label} ({about})')
 
             tk = ThesaurusKeyword()
             tk.thesaurus = thesaurus
@@ -147,7 +147,7 @@ def load_thesaurus(self, input_file, name, store):
                 lang = pref_label.attrib.get(LANG_ATTRIB)
                 label = pref_label.text
 
-                print('    Label {}: {}'.format(lang, label))
+                print(f'    Label {lang}: {label}')
 
                 tkl = ThesaurusKeywordLabel()
                 tkl.keyword = tk
@@ -174,18 +174,20 @@ def create_fake_thesaurus(self, name):
             tk.alt_label = keyword + '_alt'
             tk.save()
 
-            for l in ['it', 'en', 'es']:
+            for _l in ['it', 'en', 'es']:
                 tkl = ThesaurusKeywordLabel()
                 tkl.keyword = tk
-                tkl.lang = l
-                tkl.label = keyword + "_l_" + l + "_t_" + name
+                tkl.lang = _l
+                tkl.label = keyword + "_l_" + _l + "_t_" + name
                 tkl.save()
 
+
 def get_all_lang_available_with_title(items: List, LANG_ATTRIB: str):
     return [(item.attrib.get(LANG_ATTRIB), item.text) for item in items]
 
+
 def determinate_value(available_lang: List, default_lang: str):
-    sorted_lang = sorted(available_lang, key= lambda lang: '' if lang[0] is None else lang[0])
+    sorted_lang = sorted(available_lang, key=lambda lang: '' if lang[0] is None else lang[0])
     for item in sorted_lang:
         if item[0] is None:
             return item[1]

geonode/base/management/commands/migrate_baseurl.py

@@ -74,51 +74,51 @@ def handle(self, **options):
         if not target_address or len(target_address) == 0:
             raise CommandError("Target Address '--target-address' is mandatory")
 
-        print("This will change all Maps, Layers, \
-Styles and Links Base URLs from [%s] to [%s]." % (source_address, target_address))
+        print(f"This will change all Maps, Layers, \
+Styles and Links Base URLs from [{source_address}] to [{target_address}].")
         print("The operation may take some time, depending on the amount of Layer on GeoNode.")
         message = 'You want to proceed?'
 
         if force_exec or helpers.confirm(prompt=message, resp=False):
             try:
                 _cnt = Map.objects.filter(thumbnail_url__icontains=source_address).update(
                     thumbnail_url=Func(
-                        F('thumbnail_url'),Value(source_address),Value(target_address),function='replace'))
-                logger.info("Updated %s Maps" % _cnt)
+                        F('thumbnail_url'), Value(source_address), Value(target_address), function='replace'))
+                logger.info(f"Updated {_cnt} Maps")
 
                 _cnt = MapLayer.objects.filter(ows_url__icontains=source_address).update(
                     ows_url=Func(
-                        F('ows_url'),Value(source_address),Value(target_address),function='replace'))
+                        F('ows_url'), Value(source_address), Value(target_address), function='replace'))
                 MapLayer.objects.filter(layer_params__icontains=source_address).update(
                     layer_params=Func(
-                        F('layer_params'),Value(source_address),Value(target_address),function='replace'))
-                logger.info("Updated %s MapLayers" % _cnt)
+                        F('layer_params'), Value(source_address), Value(target_address), function='replace'))
+                logger.info(f"Updated {_cnt} MapLayers")
 
                 _cnt = Layer.objects.filter(thumbnail_url__icontains=source_address).update(
                     thumbnail_url=Func(
-                        F('thumbnail_url'),Value(source_address),Value(target_address),function='replace'))
-                logger.info("Updated %s Layers" % _cnt)
+                        F('thumbnail_url'), Value(source_address), Value(target_address), function='replace'))
+                logger.info(f"Updated {_cnt} Layers")
 
                 _cnt = Style.objects.filter(sld_url__icontains=source_address).update(
                     sld_url=Func(
-                        F('sld_url'),Value(source_address),Value(target_address),function='replace'))
-                logger.info("Updated %s Styles" % _cnt)
+                        F('sld_url'), Value(source_address), Value(target_address), function='replace'))
+                logger.info(f"Updated {_cnt} Styles")
 
                 _cnt = Link.objects.filter(url__icontains=source_address).update(
                     url=Func(
-                        F('url'),Value(source_address),Value(target_address),function='replace'))
-                logger.info("Updated %s Links" % _cnt)
+                        F('url'), Value(source_address), Value(target_address), function='replace'))
+                logger.info(f"Updated {_cnt} Links")
 
                 _cnt = ResourceBase.objects.filter(thumbnail_url__icontains=source_address).update(
                     thumbnail_url=Func(
-                        F('thumbnail_url'),Value(source_address),Value(target_address),function='replace'))
+                        F('thumbnail_url'), Value(source_address), Value(target_address), function='replace'))
                 _cnt += ResourceBase.objects.filter(csw_anytext__icontains=source_address).update(
                     csw_anytext=Func(
                         F('csw_anytext'), Value(source_address), Value(target_address), function='replace'))
                 _cnt += ResourceBase.objects.filter(metadata_xml__icontains=source_address).update(
                     metadata_xml=Func(
                         F('metadata_xml'), Value(source_address), Value(target_address), function='replace'))
-                logger.info("Updated %s ResourceBases" % _cnt)
+                logger.info(f"Updated {_cnt} ResourceBases")
 
                 site = Site.objects.get_current()
                 if site:
@@ -132,7 +132,7 @@ def handle(self, **options):
                         _cnt = Application.objects.filter(name='GeoServer').update(
                             redirect_uris=Func(
                                 F('redirect_uris'), Value(source_address), Value(target_address), function='replace'))
-                        logger.info("Updated %s OAUth2 Redirect URIs" % _cnt)
+                        logger.info(f"Updated {_cnt} OAUth2 Redirect URIs")
 
             finally:
                 print("...done!")

geonode/base/management/commands/migrate_notifications.py

@@ -43,7 +43,7 @@ def handle(self, **options):
         try:
             c.execute('select medium, send, notice_type_id, user_id from notification_noticesetting;')
         except ProgrammingError as err:
-            log.error("No table for notification app, exiting")
+            log.error(f"No table for notification app, exiting: {err}")
             # no source of data, bye!
             return
 

geonode/base/management/commands/set_all_layers_alternate.py

@@ -66,15 +66,13 @@ def handle(self, *args, **options):
             all_layers = all_layers.filter(owner__username=username)
 
         for index, layer in enumerate(all_layers):
-            logger.info("[%s / %s] Checking 'alternate' of Layer [%s] ..." % ((index + 1), len(all_layers), layer.name))
+            logger.info(f"[{(index + 1)} / {len(all_layers)}] Checking 'alternate' of Layer [{layer.name}] ...")
             try:
                 if not layer.alternate:
                     layer.alternate = layer.typename
                     layer.save()
             except Exception as e:
-                # import traceback
-                # traceback.print_exc()
                 if ignore_errors:
-                    logger.error("[ERROR] Layer [%s] couldn't be updated" % (layer.name))
+                    logger.error(f"[ERROR] Layer [{layer.name}] couldn't be updated")
                 else:
                     raise e

geonode/base/management/commands/set_all_layers_metadata.py

@@ -23,13 +23,8 @@
 from geonode.layers.models import Layer
 from geonode import geoserver  # noqa
 from geonode.catalogue.models import catalogue_post_save
-from geonode.geoserver.helpers import ogc_server_settings
 import logging
 
-
-logger = logging.getLogger(__name__)
-
-
 from geonode.utils import (
     check_ogc_backend,
     set_resource_default_links
@@ -42,6 +37,8 @@
 if check_ogc_backend(geoserver.BACKEND_PACKAGE):
     from geonode.geoserver.helpers import set_attributes_from_geoserver as set_attributes
 
+logger = logging.getLogger(__name__)
+
 
 class Command(BaseCommand):
     help = 'Resets Metadata Attributes and Schema to All Layers'
@@ -110,7 +107,7 @@ def handle(self, *args, **options):
             all_layers = all_layers.filter(owner__username=username)
 
         for index, layer in enumerate(all_layers):
-            print("[%s / %s] Updating Layer [%s] ..." % ((index + 1), len(all_layers), layer.name))
+            print(f"[{(index + 1)} / {len(all_layers)}] Updating Layer [{layer.name}] ...")
             try:
                 # recalculate the layer statistics
                 set_attributes(layer, overwrite=True)
@@ -128,7 +125,7 @@ def handle(self, *args, **options):
                 import traceback
                 traceback.print_exc()
                 if ignore_errors:
-                    logger.error("[ERROR] Layer [%s] couldn't be updated" % (layer.name))
+                    logger.error(f"[ERROR] Layer [{layer.name}] couldn't be updated")
                 else:
                     raise e
 

geonode/base/management/commands/set_all_layers_public.py

@@ -36,7 +36,7 @@ def handle(self, *args, **options):
         all_layers = Layer.objects.all()
 
         for index, layer in enumerate(all_layers):
-            print("[%s / %s] Setting public permissions to Layer [%s] ..." % ((index + 1), len(all_layers), layer.name))
+            print(f"[{(index + 1)} / {len(all_layers)}] Setting public permissions to Layer [{layer.name}] ...")
             try:
                 use_geofence = settings.OGC_SERVER['default'].get(
                     "GEOFENCE_SECURITY_ENABLED", False)
@@ -53,4 +53,4 @@ def handle(self, *args, **options):
                 perm_spec["users"]["AnonymousUser"] = ['view_resourcebase', 'download_resourcebase']
                 layer.set_permissions(perm_spec)
             except Exception:
-                logger.error("[ERROR] Layer [%s] couldn't be updated" % (layer.name))
+                logger.error(f"[ERROR] Layer [{layer.name}] couldn't be updated")