style: fix linter issues

Author: guilhermearaujo

flamingo/models/app.py

@@ -48,7 +48,7 @@ class App(Document):
     _environment: Environment = None
 
     def __post_init__(self):
-        environment = self.environment  # check if environment name actually exists, and caches it
+        _ = self.environment  # check if environment name actually exists, and caches it
 
         self.name = slugify(self.name)
 
@@ -117,7 +117,12 @@ def get_all_env_vars(self) -> List[EnvVar]:
 
         if self.gateway:
             all_vars.extend([
-                EnvVar(key='GCP_GATEWAY_ENDPOINT', value=self.gateway.gateway_endpoint, is_secret=False, source=by_flamingo),
+                EnvVar(
+                    key='GCP_GATEWAY_ENDPOINT',
+                    value=self.gateway.gateway_endpoint,
+                    is_secret=False,
+                    source=by_flamingo,
+                ),
             ])
 
         for integrated_app in self.integrated_apps:

flamingo/models/build.py

@@ -38,8 +38,7 @@ class Build(EmbeddedDocument):
     def __post_init__(self):
         if not self.deploy_tag and not self.deploy_branch:
             raise exceptions.ValidationError(message="Either deploy_tag or deploy_branch must be provided")
-        if self.max_instances < 1:
-            self.max_instances = 1
+        self.max_instances = max(self.max_instances, 1)
 
     def serialize(self) -> dict:
         data = super().serialize()

flamingo/services/alias_engine.py

@@ -33,7 +33,7 @@ def replace(self, virtual_value):
         for alias_to in aliases_to:
             try:
                 replace_with = self.replacements[alias_to]
-                new_value = new_value.replace("${%s}" % alias_to, replace_with)
+                new_value = new_value.replace(f"${{{alias_to}}}", replace_with)
             except KeyError as e:
                 raise ValidationError(f"Could not find the referenced value for {alias_to}") from e
 
@@ -43,8 +43,8 @@ def replace(self, virtual_value):
 class AliasEngine:
     def __init__(self, items: KeyValue, replacements: ReplacementEngine = None):
         super().__init__()
-        self._concrete: KeyValue = dict()
-        self._virtual: KeyValue = dict()
+        self._concrete: KeyValue = {}
+        self._virtual: KeyValue = {}
 
         if not replacements:
             replacements = ReplacementEngine()

flamingo/services/bootstrap.py

@@ -1,6 +1,5 @@
 from dataclasses import dataclass
 
-import settings
 from models.app import App
 from models.base import random_password
 from models.bucket import Bucket

flamingo/services/builders.py

@@ -85,14 +85,14 @@ def _get_db_as_param(self, command: str) -> List[str]:
 
     def _get_env_var_as_param(self, command: str = '--set-env-var') -> List[str]:
         params = []
-        for key, value in self._env_vars.items():
+        for key, _ in self._env_vars.items():
             sub_variable = getattr(self._substitution, f'{self.ENV_PREFIX_KEY}{key}')
             params.extend([command, sub_variable.as_env_var(key=key)])
         return params
 
     def _get_build_args_as_param(self, command: str = '--build-arg') -> List[str]:
         build_params = []
-        for key, value in self._build_args.items():
+        for key, _ in self._build_args.items():
             sub_variable = getattr(self._substitution, key)
             build_params.extend([command, sub_variable.as_env_var()])
         return build_params
@@ -414,14 +414,14 @@ def get_url(self):
 class CloudFunctionsFactory(BuildTriggerFactory):
     # TODO: setup this <https://cloud.google.com/functions/docs/reference/iam/roles#additional-configuration>
     def _get_setup_params(self) -> KeyValue:
-        from gcp_pilot.functions import CloudFunctions
+        from gcp_pilot.functions import CloudFunctions  # pylint: disable=import-outside-toplevel
 
         if self._build.deploy_tag:
             kwargs = dict(tag=self._build.deploy_tag)
         elif self._build.deploy_branch:
             kwargs = dict(branch=self._build.deploy_branch)
         else:
-            kwargs = dict()
+            kwargs = {}
 
         directory = self._build.directory
         repo_url = CloudFunctions.build_repo_source(

flamingo/services/foundations.py

@@ -63,7 +63,6 @@ def get_jobs(self) -> Dict[str, Callable]:
         }
 
     async def setup_iam(self):
-        iam = IdentityAccessManager()
         grm = ResourceManager()
 
         roles = [
@@ -158,7 +157,7 @@ async def setup_placeholder(self):
                 )
 
             try:
-                gateway_config = gateway.create_config(
+                gateway.create_config(
                     config_name=f"{self.app.name}-placeholder",
                     api_name=self.app.gateway.api_name,
                     service_account=self.app.service_account.email,
@@ -167,7 +166,7 @@ async def setup_placeholder(self):
                     project_id=self.app.project.id,
                 )
             except AlreadyExists:
-                gateway_config = gateway.get_config(
+                gateway.get_config(
                     config_name=f"{self.app.name}-placeholder",
                     api_name=self.app.gateway.api_name,
                     project_id=self.app.project.id,

flamingo/views/app_views.py

@@ -3,13 +3,13 @@
 from sanic import Blueprint
 from sanic.request import Request
 from sanic_rest import exceptions
+from sanic_rest.views import NestedListView, DetailView, ListView, ResponseType
 
 from models.app import App
 from models.database import Database
 from models.env_var import EnvVar
 from services.bootstrap import AppBootstrap
 from services.foundations import AppFoundation
-from sanic_rest.views import NestedListView, DetailView, ListView, ResponseType
 
 apps = Blueprint('apps', url_prefix='/apps')
 
@@ -25,34 +25,40 @@ class AppDetailView(DetailView):
 class AppBoostrapView(NestedListView):
     nest_model = App
 
-    async def perform_get(self, request: Request, obj: App) -> ResponseType:
-        bootstrap = AppBootstrap(app=obj)
+    async def perform_get(self, request: Request, nest_obj: App) -> ResponseType:
+        bootstrap = AppBootstrap(app=nest_obj)
         changes = bootstrap.check()
         return changes, 200
 
-    async def perform_post(self, request: Request, obj: App) -> ResponseType:
-        bootstrap = AppBootstrap(app=obj)
+    async def perform_post(self, request: Request, nest_obj: App) -> ResponseType:
+        bootstrap = AppBootstrap(app=nest_obj)
         new_obj = bootstrap.apply()
         return new_obj.serialize(), 200
 
-    async def perform_delete(self, request: Request, obj: App) -> ResponseType:
+    async def perform_put(self, request: Request, nest_obj: App) -> ResponseType:
+        raise exceptions.NotAllowedError()
+
+    async def perform_delete(self, request: Request, nest_obj: App) -> ResponseType:
         raise exceptions.NotAllowedError()
 
 
 class AppInitializeView(NestedListView):
     nest_model = App
 
-    async def perform_get(self, request: Request, obj: App) -> ResponseType:
-        foundation = AppFoundation(app=obj)
+    async def perform_get(self, request: Request, nest_obj: App) -> ResponseType:
+        foundation = AppFoundation(app=nest_obj)
         jobs = foundation.get_jobs()
         return {'jobs': list(jobs.keys())}, 200
 
-    async def perform_post(self, request: Request, obj: App) -> ResponseType:
-        foundation = AppFoundation(app=obj)
+    async def perform_post(self, request: Request, nest_obj: App) -> ResponseType:
+        foundation = AppFoundation(app=nest_obj)
         jobs = foundation.build()
         return {'jobs': jobs}, 202
 
-    async def perform_delete(self, request: Request, obj: App) -> ResponseType:
+    async def perform_put(self, request: Request, nest_obj: App) -> ResponseType:
+        raise exceptions.NotAllowedError()
+
+    async def perform_delete(self, request: Request, nest_obj: App) -> ResponseType:
         raise exceptions.NotAllowedError()
 
 
@@ -83,6 +89,9 @@ async def perform_post(self, request: Request, nest_obj: App) -> ResponseType:
         }
         return payload, 201
 
+    async def perform_put(self, request: Request, nest_obj: App) -> ResponseType:
+        raise exceptions.NotAllowedError()
+
     async def perform_delete(self, request: Request, nest_obj: App) -> ResponseType:
         for key in request.json:
             nest_obj.unset_env_var(key=key)
@@ -109,6 +118,9 @@ async def perform_post(self, request: Request, nest_obj: App) -> ResponseType:
         payload = new_obj.database.serialize()
         return payload, 201
 
+    async def perform_put(self, request: Request, nest_obj: App) -> ResponseType:
+        raise exceptions.NotAllowedError()
+
     async def perform_delete(self, request: Request, nest_obj: App) -> ResponseType:
         nest_obj.database = None
         nest_obj.save()
@@ -133,7 +145,10 @@ async def perform_post(self, request: Request, nest_obj: App) -> ResponseType:
         # TODO Add support to re-deploy
         return {'trigger_id': trigger_id}, 201
 
-    async def perform_delete(self, request: Request, obj: App) -> ResponseType:
+    async def perform_put(self, request: Request, nest_obj: App) -> ResponseType:
+        raise exceptions.NotAllowedError()
+
+    async def perform_delete(self, request: Request, nest_obj: App) -> ResponseType:
         raise exceptions.NotAllowedError()
 
 

flamingo/views/build_pack_views.py

@@ -1,8 +1,8 @@
 from sanic import Blueprint
 from sanic.request import File
+from sanic_rest.views import DetailView, ListView, PayloadType
 
 from models.buildpack import BuildPack
-from sanic_rest.views import DetailView, ListView, PayloadType
 
 build_packs = Blueprint('build-packs', url_prefix='/build-packs')
 

flamingo/views/environment_views.py

@@ -1,10 +1,10 @@
 from sanic import Blueprint
 from sanic.request import Request
 from sanic_rest import exceptions
+from sanic_rest.views import DetailView, ListView, NestedListView, ResponseType
 
 from models.environment import Environment
 from services.foundations import EnvironmentFoundation
-from sanic_rest.views import DetailView, ListView, NestedListView, ResponseType
 
 environments = Blueprint('environments', url_prefix='/environments')
 
@@ -30,7 +30,10 @@ async def perform_post(self, request: Request, nest_obj: Environment) -> Respons
         jobs = foundation.build()
         return {'jobs': jobs}, 202
 
-    async def perform_delete(self, request: Request, obj: Environment) -> ResponseType:
+    async def perform_put(self, request: Request, nest_obj: Environment) -> ResponseType:
+        raise exceptions.NotAllowedError()
+
+    async def perform_delete(self, request: Request, nest_obj: Environment) -> ResponseType:
         raise exceptions.NotAllowedError()
 
 

flamingo/views/hook_views.py

@@ -1,6 +1,6 @@
 import logging
 from datetime import datetime, timezone
-from typing import Dict
+from typing import Dict, Optional
 
 from gcp_pilot.datastore import DoesNotExist, MultipleObjectsFound
 from gcp_pilot.pubsub import Message
@@ -59,14 +59,15 @@ async def post(self, request: Request) -> HTTPResponse:
             return json({'error': f"Failed handling build hook: {e}", 'payload': payload}, 400)
         return json({'status': 'done'}, 202)
 
-    def _get_timestamp(self, payload: Dict) -> datetime:
+    def _get_timestamp(self, payload: Dict) -> Optional[datetime]:
         time_fields = ['finishTime', 'startTime', 'createTime']
         for time_field in time_fields:
             try:
                 date_str = payload[time_field]
                 return datetime.strptime(date_str.split('.')[0], '%Y-%m-%dT%H:%M:%S').astimezone(tz=timezone.utc)
             except KeyError:
                 continue
+        return None
 
     def _get_app(self, trigger_id: str) -> App:
         return App.documents.get(build__trigger_id=trigger_id)