do: fix pylint warnings

Author: ImadYIdrissi

engine/data_pipelines/common/auth/gcloud.py

@@ -8,11 +8,11 @@
 def get_auth_token() -> str:
     """Retrieve a Google Cloud SDK authentication token.
 
-    Returns: Authentication token.
+    Returns:
+        str: Authentication token.
 
     Raises:
-      subprocess.CalledProcessError: If authentication fails.
-
+        subprocess.CalledProcessError: If authentication fails.
     """
     try:
         result = subprocess.run(  # nosec
@@ -28,9 +28,9 @@ def get_auth_token() -> str:
         if token:
             LOGGER.info("Successfully authenticated with Google Cloud SDK.")
             return token
-        else:
-            LOGGER.error("No token returned during authentication.")
-            raise subprocess.CalledProcessError(returncode=1, cmd=result.args, stderr="No token returned")
+
+        LOGGER.error("No token returned during authentication.")
+        raise subprocess.CalledProcessError(returncode=1, cmd=result.args, stderr="No token returned")
 
     except subprocess.CalledProcessError as e:
         msgs = e.stderr.split("\n")

engine/data_pipelines/microservices/google_analytics_pipeline/main_pipeline.py

@@ -1,7 +1,6 @@
 """Main pipeline module for google_analytics_pipeline."""
 
 import os
-from google.cloud import bigquery
 from google.auth import default
 from engine.data_pipelines.microservices.google_analytics_pipeline import RENAMER_SESSIONS_HITS, QUERY_SESSIONS_HITS
 from engine.data_pipelines.common.logging import LOGGER
@@ -26,16 +25,13 @@ def main():
                 "Cloud Run has the correct permissions."
             ) from e
         LOGGER.info("Using GOOGLE_APPLICATION_CREDENTIALS for authentication.")
-        credentials = None  # Automatically handled by the environment
-
-    # Initialize BigQuery client
-    client = bigquery.Client(credentials=credentials, project=project)
+        credentials = None  # Automatically handled by the environment*
 
     LOGGER.info("Extract data.")
     df = read_from_bigquery(
         query=QUERY_SESSIONS_HITS,
         project_id=project,
-        credentials=client._credentials,  # Pass client credentials
+        credentials=credentials,  # Pass client credentials
     )
 
     LOGGER.info("Transform - Staging")

pyproject.toml

@@ -1,5 +1,5 @@
 [tool.black]
-line-length=119
+line-length = 119
 
 [tool.flake8]
 max-line-length = 119
@@ -9,4 +9,11 @@ ignore = ["E203"]
 max-line-length = 119
 
 [tool.pylint.design]
-max-args = 10  # Maximum number of arguments for a func
+max-args = 10  # Maximum number of arguments for a function
+
+[tool.pylint.messages_control]
+disable = [
+    "logging-fstring-interpolation",  # Ignore logging f-string interpolation warnings
+    "too-many-positional-arguments",  # Disable positional arguments limit warning
+    "broad-exception-caught" # Disable board exception warnings
+]

services/public_apis/fastAPI_hello_world/__init__.py renamed to services/public_apis/fast_api_hello_world/__init__.py

@@ -38,9 +38,9 @@ async def read_user(user_id: int):
 class ModelName(str, Enum):
     """Enumeration for machine learning models."""
 
-    alexnet = "alexnet"
-    resnet = "resnet"
-    lenet = "lenet"
+    ALEXNET = "alexnet"
+    RESNET = "resnet"
+    LENET = "lenet"
 
 
 @app.get("/models/{model_name}")
@@ -53,7 +53,7 @@ async def get_model(model_name: ModelName):
     Returns:
         dict: Details about the selected model.
     """
-    if model_name is ModelName.alexnet:
+    if model_name is ModelName.ALEXNET:
         return {"model_name": model_name, "message": "Deep Learning FTW!"}
 
     if model_name.value == "lenet":
@@ -178,11 +178,11 @@ async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 10
     return {"q": q, "skip": skip, "limit": limit}
 
 
-commonDeps = Annotated[dict, Depends(common_parameters)]
+CommonDeps = Annotated[dict, Depends(common_parameters)]
 
 
 @app.get("/shmushes/")
-async def read_smushes(commons: commonDeps):
+async def read_smushes(commons: CommonDeps):
     """Read shmushes with common parameters.
 
     Args:
@@ -195,7 +195,7 @@ async def read_smushes(commons: commonDeps):
 
 
 @app.get("/shmooleys/")
-async def read_shmooleys(commons: commonDeps):
+async def read_shmooleys(commons: CommonDeps):
     """Read shmooleys with common parameters.
 
     Args:

services/public_apis/fastAPI_hello_world/main.py renamed to services/public_apis/fast_api_hello_world/main.py

@@ -27,5 +27,5 @@ def get_agg_data():
 
 if __name__ == "__main__":
     # Serve the FastAPI app on the port provided by the environment variable
-    port = int(os.getenv("PORT", 8000))
+    port = int(os.getenv("PORT", "8000"))
     uvicorn.run(app, host="0.0.0.0", port=port)  # nosec