lint and format strings

Author: madkote

fastapi_plugins/_redis.py

@@ -67,13 +67,9 @@ def get_redis_address(self) -> str:
         if self.redis_url:
             return self.redis_url
         elif self.redis_db:
-            return 'redis://%s:%s/%s' % (
-                self.redis_host,
-                self.redis_port,
-                self.redis_db
-            )
+            return f'redis://{self.redis_host}:{self.redis_port}/{self.redis_db}'
         else:
-            return 'redis://%s:%s' % (self.redis_host, self.redis_port)
+            return f'redis://{self.redis_host}:{self.redis_port}'
 
     # TODO: xxx the customer validator does not work
     def get_sentinels(self) -> typing.List:
@@ -88,11 +84,7 @@ def get_sentinels(self) -> typing.List:
                     if _conn.strip()
                 ]
             except Exception as e:
-                raise RuntimeError(
-                    'bad sentinels string :: %s :: %s :: %s' % (
-                        type(e), str(e), self.redis_sentinels
-                    )
-                )
+                raise RuntimeError(f'bad sentinels string :: {type(e)} :: {str(e)} :: {self.redis_sentinels}')  # noqa
         else:
             return []
 
@@ -101,7 +93,7 @@ class RedisPlugin(Plugin, ControlHealthMixin):
     DEFAULT_CONFIG_CLASS = RedisSettings
 
     def _on_init(self) -> None:
-        self.redis: typing.Union[aioredis.Redis, aioredis_sentinel.Sentinel] = None # noqa E501
+        self.redis: typing.Union[aioredis.Redis, aioredis_sentinel.Sentinel] = None
 
     async def _on_call(self) -> typing.Any:
         if self.redis is None:
@@ -114,9 +106,7 @@ async def _on_call(self) -> typing.Any:
         elif self.config.redis_type == RedisType.fakeredis:
             conn = self.redis
         else:
-            raise NotImplementedError(
-                'Redis type %s is not implemented' % self.config.redis_type
-            )
+            raise NotImplementedError(f'Redis type {self.config.redis_type} is not implemented')    # noqa
         #
         conn.TTL = self.config.redis_ttl
         return conn
@@ -163,9 +153,7 @@ async def init(self):
             address = self.config.get_sentinels()
             method = aioredis_sentinel.Sentinel
         else:
-            raise NotImplementedError(
-                'Redis type %s is not implemented' % self.config.redis_type
-            )
+            raise NotImplementedError(f'Redis type {self.config.redis_type} is not implemented')    # noqa
         #
         if not address:
             raise ValueError('Redis address is empty')
@@ -202,11 +190,9 @@ async def ping(self):
         elif self.config.redis_type == RedisType.fakeredis:
             return await self.redis.ping()
         elif self.config.redis_type == RedisType.sentinel:
-            return await self.redis.master_for(self.config.redis_sentinel_master).ping()    # noqa E501
+            return await self.redis.master_for(self.config.redis_sentinel_master).ping()
         else:
-            raise NotImplementedError(
-                'Redis type %s.ping() is not implemented' % self.config.redis_type          # noqa E501
-            )
+            raise NotImplementedError(f'Redis type {self.config.redis_type}.ping() is not implemented') # noqa
 
 
 redis_plugin = RedisPlugin()

fastapi_plugins/logger.py

@@ -72,7 +72,7 @@ def format_line(self, extra: typing.Dict) -> str:
         outarr = []
         for k, v in extra.items():
             if v is None:
-                outarr.append('%s=' % k)
+                outarr.append(f'{k}=')
                 continue
             if isinstance(v, bool):
                 v = 'true' if v else 'false'
@@ -82,15 +82,15 @@ def format_line(self, extra: typing.Dict) -> str:
                 if isinstance(v, (dict, object)):
                     v = str(v)
                 v = '"%s"' % v.replace('"', '\\"')
-            outarr.append('%s=%s' % (k, v))
+            outarr.append(f'{k}={v}')
         return ' '.join(outarr)
 
     def format(self, record):
         return ' '.join(
             [
-                'at=%s' % record.levelname,
+                f'at={record.levelname}',
                 'msg="%s"' % record.getMessage().replace('"', '\\"'),
-                'process=%s' % record.processName,
+                f'process={record.processName}',
                 self.format_line(getattr(record, 'context', {})),
             ]
         ).strip()

fastapi_plugins/memcached.py

@@ -94,9 +94,7 @@ async def _init_memcached():
         try:
             await _init_memcached()
         except Exception as e:
-            raise MemcachedError(
-                'Memcached initialization failed :: %s :: %s' % (type(e), e)
-            )
+            raise MemcachedError(f'Memcached initialization failed :: {type(e)} :: {str(e)}')   # noqa
 
     async def terminate(self):
         self.config = None
@@ -122,4 +120,4 @@ async def depends_memcached(
     return await conn.app.state.MEMCACHED()
 
 
-TMemcachedPlugin = Annotated[MemcachedClient, fastapi.Depends(depends_memcached)]   # noqa E501
+TMemcachedPlugin = Annotated[MemcachedClient, fastapi.Depends(depends_memcached)]

fastapi_plugins/scheduler.py

@@ -23,7 +23,7 @@
 ]
 __author__ = 'madkote <madkote(at)bluewin.ch>'
 __version__ = '.'.join(str(x) for x in VERSION)
-__copyright__ = 'Copyright 2021, madkote'
+__copyright__ = 'Copyright 2025, madkote'
 
 
 class SchedulerError(PluginError):
@@ -139,4 +139,4 @@ async def depends_scheduler(
     return await conn.app.state.AIOJOBS_SCHEDULER()
 
 
-TSchedulerPlugin = Annotated[aiojobs.Scheduler, fastapi.Depends(depends_scheduler)] # noqa E501
+TSchedulerPlugin = Annotated[aiojobs.Scheduler, fastapi.Depends(depends_scheduler)]

fastapi_plugins/settings.py

@@ -34,7 +34,7 @@
 ]
 __author__ = 'madkote <madkote(at)bluewin.ch>'
 __version__ = '.'.join(str(x) for x in VERSION)
-__copyright__ = 'Copyright 2021, madkote RES'
+__copyright__ = 'Copyright 2025, madkote RES'
 
 DEFAULT_CONFIG_ENVVAR: str = 'CONFIG_NAME'
 DEFAULT_CONFIG_NAME: str = 'docker'
@@ -65,9 +65,9 @@ def reset(self) -> None:
 
     def get_config(
             self,
-            config_or_name: typing.Union[str, pydantic_settings.BaseSettings]=None, # noqa E501
-            config_name_default: str=DEFAULT_CONFIG_NAME,
-            config_name_envvar: str=DEFAULT_CONFIG_ENVVAR
+            config_or_name: typing.Union[str, pydantic_settings.BaseSettings] = None,
+            config_name_default: str = DEFAULT_CONFIG_NAME,
+            config_name_envvar: str = DEFAULT_CONFIG_ENVVAR
     ) -> pydantic_settings.BaseSettings:
         if isinstance(config_or_name, pydantic_settings.BaseSettings):
             return config_or_name
@@ -80,7 +80,7 @@ def get_config(
             default=config_or_name
         )
         if config_or_name not in self._settings_map:
-            raise ConfigError('Unknown configuration "%s"' % config_or_name)
+            raise ConfigError(f'Unknown configuration "{config_or_name}"')
         return self._settings_map[config_or_name]()
 
 
@@ -190,4 +190,4 @@ async def depends_config(
     return await conn.app.state.PLUGIN_CONFIG()
 
 
-TConfigPlugin = Annotated[pydantic_settings.BaseSettings, fastapi.Depends(depends_config)]   # noqa E501
+TConfigPlugin = Annotated[pydantic_settings.BaseSettings, fastapi.Depends(depends_config)]