lint

.pylintrc

@@ -0,0 +1,5 @@
+[pycodestyle]
+max-line-length = 140
+
+[pylint.messages control]
+disable = missing-module-docstring, missing-function-docstring, missing-class-docstring

pystructurizr/cli.py

@@ -1,19 +1,21 @@
-import click
 import asyncio
+import json
 import os
 import shutil
 import subprocess
-import json
-from .cli_helper import generate_diagram_code, generate_diagram_code_in_child_process, generate_svg, ensure_tmp_folder_exists
-from .cli_watcher import observe_modules
-from .cloudstorage import create_cloud_storage, CloudStorage
 
+import click
+
+from .cli_helper import (ensure_tmp_folder_exists, generate_diagram_code,
+                         generate_diagram_code_in_child_process, generate_svg)
+from .cli_watcher import observe_modules
+from .cloudstorage import CloudStorage, create_cloud_storage
 
 
 @click.command()
 @click.option('--view', prompt='Your view file (e.g. example.componentview)',
               help='The view file to generate.')
-@click.option('--as-json', is_flag=True, default=False, 
+@click.option('--as-json', is_flag=True, default=False,
               help='Dumps the generated code and the imported modules as a json object')
 def dump(view, as_json):
     diagram_code, imported_modules = generate_diagram_code(view)
@@ -47,6 +49,7 @@ async def async_behavior():
     async def observe_loop():
         modules_to_watch = await async_behavior()
         click.echo("Launching webserver...")
+        # pylint: disable=consider-using-with
         subprocess.Popen(f"httpwatcher --root {tmp_folder} --watch {tmp_folder}", shell=True)
         await observe_modules(modules_to_watch, async_behavior)
 
@@ -76,7 +79,7 @@ async def async_behavior():
         cloud_storage = create_cloud_storage(CloudStorage.Provider.GCS, gcs_credentials)
         svg_file_url = cloud_storage.upload_file(svg_file_path, bucket_name, object_name)
         print(svg_file_url)
-    
+
     asyncio.run(async_behavior())
 
 

pystructurizr/cli_helper.py

@@ -1,13 +1,12 @@
-import click
 import importlib
+import json
+import os
 import subprocess
 import sys
+
 import aiofiles
 import click
 import httpx
-import sys
-import os
-import json
 
 
 def generate_diagram_code(view: str) -> str:
@@ -18,8 +17,10 @@ def generate_diagram_code(view: str) -> str:
         code = module.workspace.dump()
         return code, imported_modules
     except ModuleNotFoundError:
+        # pylint: disable=raise-missing-from
         raise click.BadParameter("Invalid view name. Make sure you don't include the .py file extension.")
     except AttributeError:
+        # pylint: disable=raise-missing-from
         raise click.BadParameter("Non-compliant view file: make sure it exports the PyStructurizr workspace.")
 
 
@@ -35,7 +36,7 @@ def run_child_process():
 
 
 async def generate_svg(diagram_code: str, tmp_folder: str) -> str:
-    url = f"https://kroki.io/structurizr/svg"
+    url = "https://kroki.io/structurizr/svg"
     async with httpx.AsyncClient() as client:
         resp = await client.post(url, data=diagram_code)
 
@@ -44,10 +45,10 @@ async def generate_svg(diagram_code: str, tmp_folder: str) -> str:
         if resp.content:
             print(resp.content.decode())
         raise click.ClickException("Failed to create diagram")
-    
+
     svg_file_path = f"{tmp_folder}/diagram.svg"
-    async with aiofiles.open(svg_file_path, "w") as f:
-        await f.write(resp.text)
+    async with aiofiles.open(svg_file_path, "w") as svg_file:
+        await svg_file.write(resp.text)
 
     return svg_file_path
 

pystructurizr/cli_watcher.py

@@ -1,8 +1,9 @@
+import datetime
 import os
 import time
-from watchdog.observers import Observer
+
 from watchdog.events import FileSystemEventHandler
-import datetime
+from watchdog.observers import Observer
 
 
 def formatted_timestamp():

pystructurizr/cloudstorage.py

@@ -1,9 +1,14 @@
+import json
 from abc import ABC, abstractmethod
 from enum import Enum
-import json
 from typing import Dict
 
+import boto3
+from google.cloud import storage
+from google.cloud.exceptions import GoogleCloudError
+
 
+# pylint: disable=too-few-public-methods
 class CloudStorage(ABC):
     class Provider(Enum):
         GCS = "GCS"
@@ -12,40 +17,38 @@ class Provider(Enum):
     @abstractmethod
     def upload_file(self, file_path: str, bucket_name: str, object_name: str) -> str:
         pass
- 
+
 
 class GCS(CloudStorage):
     def __init__(self, gcs_credentials):
         self.gcs_credentials = gcs_credentials
 
     def upload_file(self, file_path: str, bucket_name: str, object_name: str) -> str:
-        from google.cloud import storage
-        from google.cloud.exceptions import GoogleCloudError
         try:
             gcs_client = storage.Client.from_service_account_json(self.gcs_credentials)
             gcs_bucket = gcs_client.get_bucket(bucket_name)
             gcs_blob = gcs_bucket.blob(object_name)
             gcs_blob.upload_from_filename(file_path)
             return f"https://storage.googleapis.com/{bucket_name}/{object_name}"
-        except GoogleCloudError as e:
+        except GoogleCloudError as exc:
             print("An error occurred while uploading the file to Google Cloud Storage:")
-            print(e)
+            print(exc)
+            return ""
 
 
 class S3(CloudStorage):
     def __init__(self, credentials_file: str):
-        import boto3
         self.credentials = self._load_credentials(credentials_file)
         self.client = boto3.client(
             "s3",
             aws_access_key_id=self.credentials["access_key"],
             aws_secret_access_key=self.credentials["secret_key"],
             region_name=self.credentials["region"]
         )
-    
+
     def _load_credentials(self, credentials_file: str) -> Dict[str, str]:
-        with open(credentials_file, "r") as f:
-            credentials = json.load(f)
+        with open(credentials_file, "r", encoding='utf-8') as cred:
+            credentials = json.load(cred)
         required_keys = ["access_key", "secret_key", "region"]
         if not all(key in credentials for key in required_keys):
             raise ValueError("Invalid credentials format")
@@ -60,7 +63,6 @@ def upload_file(self, file_path: str, bucket_name: str, object_name: str) -> str
 def create_cloud_storage(provider: CloudStorage.Provider, credentials_file: str) -> CloudStorage:
     if provider == CloudStorage.Provider.GCS:
         return GCS(credentials_file)
-    elif provider == CloudStorage.Provider.S3:
+    if provider == CloudStorage.Provider.S3:
         return S3(credentials_file)
-    else:
-        raise ValueError("Invalid cloud storage provider")
+    raise ValueError("Invalid cloud storage provider")