lint fixes 1

Author: felipao-mx

agave/chalice/rest_api.py

@@ -1,5 +1,5 @@
 import mimetypes
-from typing import Any, Optional, Type, cast
+from typing import Any, Callable, NoReturn, Optional, Type, cast
 from urllib.parse import urlencode
 
 try:
@@ -18,31 +18,31 @@
 
 
 class RestApiBlueprint(Blueprint):
-    def get(self, path: str, **kwargs):
+    def get(self, path: str, **kwargs) -> Callable:
         return self.route(path, methods=['GET'], **kwargs)
 
-    def post(self, path: str, **kwargs):
+    def post(self, path: str, **kwargs) -> Callable:
         return self.route(path, methods=['POST'], **kwargs)
 
-    def patch(self, path: str, **kwargs):
+    def patch(self, path: str, **kwargs) -> Callable:
         return self.route(path, methods=['PATCH'], **kwargs)
 
-    def delete(self, path: str, **kwargs):
+    def delete(self, path: str, **kwargs) -> Callable:
         return self.route(path, methods=['DELETE'], **kwargs)
 
     @property
-    def current_user_id(self):
+    def current_user_id(self) -> str:
         return self.current_request.context['user_id']
 
     @property
-    def current_platform_id(self):
+    def current_platform_id(self) -> str:
         return self.current_request.context['platform_id']
 
     @property
-    def current_api_key_id(self):
+    def current_api_key_id(self) -> str:
         return self.current_request.context['api_key_id']
 
-    def user_id_filter_required(self):
+    def user_id_filter_required(self) -> bool:
         """
         This method is required to be implemented with your own business logic.
         You are responsible of determining when `user_id` filter is required.
@@ -51,7 +51,7 @@ def user_id_filter_required(self):
             'this method should be override'
         )  # pragma: no cover
 
-    def platform_id_filter_required(self):
+    def platform_id_filter_required(self) -> bool:
         """
         This method is required to be implemented with your own business logic.
         You are responsible of determining when `user_id` filter is required.
@@ -81,7 +81,7 @@ def retrieve_object(self, resource_class: Any, resource_id: str) -> Any:
             raise NotFoundError('Not valid id')
         return data
 
-    def validate(self, validation_type: Type[BaseModel]):
+    def validate(self, validation_type: Type[BaseModel]) -> Callable:
         """This decorator validate the request body using a
         custom pydantyc model. If validation fails return a
         BadRequest response with details
@@ -103,7 +103,7 @@ def wrapper(*args, **kwargs):
 
         return decorator
 
-    def resource(self, path: str):
+    def resource(self, path: str) -> Callable:
         """Decorator to transform a class in Chalice REST endpoints
 
         @app.resource('/my_resource')
@@ -164,7 +164,7 @@ def delete(id: str):
 
                 @copy_attributes(cls)
                 def update(id: str):
-                    params = self.current_request.json_body or dict()
+                    params = self.current_request.json_body or {}
                     try:
                         data = cls.update_validator(**params)
                     except ValidationError as e:
@@ -238,7 +238,7 @@ def query():
                     next_page = <url_for_next_items>
                 }
                 """
-                params = self.current_request.query_params or dict()
+                params = self.current_request.query_params or {}
                 try:
                     query_params = cls.query_validator(**params)
                 except ValidationError as e:
@@ -272,7 +272,7 @@ def query():
 
             def _count(filters: Q):
                 count = cls.model.objects.filter(filters).count()
-                return dict(count=count)
+                return {'count': count}
 
             def _all(query: QueryParams, filters: Q):
                 if query.limit:
@@ -302,7 +302,7 @@ def _all(query: QueryParams, filters: Q):
                     if self.platform_id_filter_required():
                         params.pop('platform_id')
                     next_page_uri = f'{path}?{urlencode(params)}'
-                return dict(items=item_dicts, next_page_uri=next_page_uri)
+                return {'items': item_dicts, 'next_page_uri': next_page_uri}
 
             return cls
 

agave/core/blueprints/decorators.py

@@ -1,7 +1,7 @@
 from typing import Any, Callable, Type
 
 
-def copy_attributes(resource: Type[Any]):
+def copy_attributes(resource: Type[Any]) -> Callable:
     """
     Copy every attached property from resource methods definition to the
     real function handler.

agave/fastapi/middlewares/error_handlers.py

@@ -21,14 +21,14 @@ async def dispatch(
         except CuencaError as exc:
             return JSONResponse(
                 status_code=exc.status_code,
-                content=dict(
-                    code=exc.code,
-                    error=str(exc),
-                ),
+                content={
+                    'code': exc.code,
+                    'error': str(exc),
+                },
             )
         except AgaveError as exc:
             return JSONResponse(
-                status_code=exc.status_code, content=dict(error=exc.error)
+                status_code=exc.status_code, content={'error': exc.error}
             )
 
 

agave/fastapi/rest_api.py

@@ -1,5 +1,5 @@
 import mimetypes
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 from urllib.parse import urlencode
 
 from cuenca_validations.types import QueryParams
@@ -83,7 +83,7 @@ async def retrieve_object(
             raise NotFoundError('Not valid id')
         return data
 
-    def resource(self, path: str):
+    def resource(self, path: str) -> Callable:
         """Decorator to transform a class in FastApi REST endpoints
 
         @app.resource('/my_resource')
@@ -111,7 +111,7 @@ def wrapper_resource_class(cls):
             :param cls: Resoucre class
             :return:
             """
-            response_model = Any
+            response_model: type[BaseModel] = None
             response_sample = {}
             include_in_schema = getattr(cls, 'include_in_schema', True)
             if hasattr(cls, 'response_model'):
@@ -399,7 +399,7 @@ async def query(
 
             async def _count(filters: Q):
                 count = await cls.model.objects.filter(filters).async_count()
-                return dict(count=count)
+                return {'count': count}
 
             async def _all(query: QueryParams, filters: Q, resource_path: str):
                 if query.limit:
@@ -431,7 +431,7 @@ async def _all(query: QueryParams, filters: Q, resource_path: str):
                     if self.platform_id_filter_required():
                         params.pop('platform_id')
                     next_page_uri = f'{resource_path}?{urlencode(params)}'
-                return dict(items=item_dicts, next_page_uri=next_page_uri)
+                return {'items': item_dicts, 'next_page_uri': next_page_uri}
 
             return cls
 

agave/tasks/sqs_tasks.py

@@ -6,7 +6,7 @@
 from functools import wraps
 from itertools import count
 from json import JSONDecodeError
-from typing import Callable
+from typing import Any, Callable, List
 
 from aiobotocore.httpsession import HTTPClientError
 from aiobotocore.session import get_session
@@ -121,7 +121,7 @@ async def message_consumer(
             yield message
 
 
-async def get_running_fast_agave_tasks():
+async def get_running_fast_agave_tasks() -> List[Any]:
     return [
         t
         for t in asyncio.all_tasks()
@@ -136,7 +136,7 @@ def task(
     visibility_timeout: int = 3600,
     max_retries: int = 1,
     max_concurrent_tasks: int = 5,
-):
+) -> Callable:
     def task_builder(task_func: Callable):
         @wraps(task_func)
         async def start_task(*args, **kwargs) -> None:

agave/tools/asyncio/sqs_client.py

@@ -37,13 +37,13 @@ async def __aexit__(
     ) -> None:
         await self.close()
 
-    async def start(self):
+    async def start(self) -> None:
         session = get_session()
         context = session.create_client('sqs', self.region_name)
         self._background_tasks = set()
         self._sqs = await context.__aenter__()
 
-    async def close(self):
+    async def close(self) -> None:
         await self._sqs.__aexit__(None, None, None)
 
     async def send_message(

agave/tools/celery.py

@@ -15,34 +15,34 @@ def build_celery_message(
     task_id = str(uuid4())
     # la definición de esta plantila se encuentra en:
     # docs.celeryproject.org/en/stable/internals/protocol.html#definition
-    message = dict(
-        properties=dict(
-            correlation_id=task_id,
-            content_type='application/json',
-            content_encoding='utf-8',
-            body_encoding='base64',
-            delivery_info=dict(exchange='', routing_key='celery'),
-        ),
-        headers=dict(
-            lang='py',
-            task=task_name,
-            id=task_id,
-            root_id=task_id,
-            parent_id=None,
-            group=None,
-        ),
-        body=_b64_encode(
+    message = {
+        'properties': {
+            'correlation_id': task_id,
+            'content_type': 'application/json',
+            'content_encoding': 'utf-8',
+            'body_encoding': 'base64',
+            'delivery_info': {'exchange': '', 'routing_key': 'celery'},
+        },
+        'headers': {
+            'lang': 'py',
+            'task': task_name,
+            'id': task_id,
+            'root_id': task_id,
+            'parent_id': None,
+            'group': None,
+        },
+        'body': _b64_encode(
             json.dumps(
                 (
                     args_,
                     kwargs_,
-                    dict(
-                        callbacks=None, errbacks=None, chain=None, chord=None
-                    ),
+                    {
+                        'callbacks': None, 'errbacks': None, 'chain': None, 'chord': None
+                    },
                 )
             )
         ),
-    )
+    }
     message['content-encoding'] = 'utf-8'
     message['content-type'] = 'application/json'
 

examples/chalice/app.py

@@ -20,5 +20,5 @@
 
 
 @app.route('/')
-def health_check():
-    return dict(greeting="I'm testing app!!!")
+def health_check() -> dict:
+    return {'greeting': "I'm testing app!!!"}

examples/chalice/blueprints/authed.py

@@ -16,7 +16,7 @@ class AuthedBlueprint(Blueprint):
 
     """
 
-    def route(self, path: str, **kwargs):
+    def route(self, path: str, **kwargs) -> Callable:
         """
         Builds route decorator with custom authentication.
         It is only a function wrapper for `Blueprint._register_handler` methods
@@ -48,12 +48,12 @@ def authed_handler(*args, **kwargs):
                 user_handler.__name__,
                 authed_handler,
                 authed_handler,
-                dict(path=path, kwargs=kwargs),
+                {'path': path, 'kwargs': kwargs},
             )
 
         return decorator
 
-    def user_id_filter_required(self):
+    def user_id_filter_required(self) -> bool:
         """
         It overrides `RestApiBlueprint.user_id_filter_required()` method.
 
@@ -74,7 +74,7 @@ def user_id_filter_required(self):
         """
         return False
 
-    def platform_id_filter_required(self):
+    def platform_id_filter_required(self) -> bool:
         """
         It overrides `RestApiBlueprint.platform_id_filter_required()` method.
         :return:

examples/chalice/resources/base.py

@@ -5,4 +5,4 @@
 
 @app.get('/healthy_auth')
 def health_auth_check() -> dict:
-    return dict(greeting="I'm authenticated and healthy !!!")
+    return {'greeting': "I'm authenticated and healthy !!!"}