Make Roboswag executable with -m option

roboswag/__init__.py

@@ -1,86 +1,2 @@
-from importlib.metadata import version
-
-import requests
-import urllib3
-
-from roboswag.auth import TokenHandler
-from roboswag.logger import Logger
-from roboswag.__main__ import run_roboswag
-
-urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
-
-try:
-    __version__ = version("roboswag")
-except Exception:  # pragma: no cover
-    pass
-
-
-class APIModel:
-    def __init__(
-        self,
-        base_url,
-        verify=False,
-        headers=None,
-        content_type="application/json",
-        proxies=None,
-        allow_redirects=True,
-        authorization=TokenHandler,
-    ):
-        # Set headers from init too (or reuse auth)
-        self.base_url = base_url
-        self.session = requests.Session()
-        self.session.verify = verify
-        self.content_type = content_type
-        if content_type is not None:
-            self.session.headers = {"Content-Type": content_type}
-        self.allow_redirects = allow_redirects
-        if proxies is not None:  # TODO urllib have autodetect proxy - allow to use it
-            self.session.proxies.update(proxies)
-        self.authorization = authorization
-        self.logger = Logger()
-        if headers is not None:
-            self.session.headers.update(headers)
-
-    def send_request(self, method, url, status=None, headers=None, body=None, query=None, **kwargs):
-        headers = self.trim_empty(headers)
-        query = self.trim_empty(query)
-        # auth = self.authorization(**kwargs) if self.authorization is not None else None
-        content_type = kwargs.get("content-type", self.content_type)
-        if content_type is not None:
-            headers["Content-Type"] = content_type
-
-        resp = self.session.request(
-            method,
-            url=self.base_url + url,
-            headers=headers,
-            json=body,
-            params=query,
-            # auth=auth,
-            allow_redirects=self.allow_redirects,
-        )
-        # TODO quiet mode
-        self.logger.log_request(resp)
-        self.logger.log_response(resp)
-        if status is not None:
-            assert resp.status_code == status, f"Expected return status: {status} but received: {resp.status_code}"
-        return resp
-
-    def post(self, *args, **kwargs):
-        # TODO handle files upload
-        return self.send_request("POST", *args, **kwargs)
-
-    def get(self, *args, **kwargs):
-        return self.send_request("GET", *args, **kwargs)
-
-    def put(self, *args, **kwargs):
-        return self.send_request("PUT", *args, **kwargs)
-
-    def delete(self, *args, **kwargs):
-        return self.send_request("DELETE", *args, **kwargs)
-
-    @staticmethod
-    def trim_empty(dictionary):
-        if dictionary is None:
-            return {}
-        return {key: value for key, value in dictionary.items() if value is not None}
+from roboswag.cli import run_roboswag
+from roboswag.model import APIModel

roboswag/__main__.py

@@ -1,27 +1,3 @@
-import argparse
-import sys
-from importlib.metadata import version
+from roboswag import run_roboswag
 
-from roboswag.generate.generate import generate
-
-try:
-    __version__ = version("roboswag")
-except Exception:  # pragma: no cover
-    pass
-
-
-def generate_cli():
-    parser = argparse.ArgumentParser()
-    parser.add_argument("-s", "--spec", metavar="SWAGGER", help="OpenAPI specification file")
-    parser.add_argument(
-        "-v", "--version", action="version", version=__version__, help="display Roboswag version and exit"
-    )
-    args = parser.parse_args()
-    if not args.spec:
-        print("Please provide specification file with '-s' option")
-        sys.exit(1)
-    generate(args.spec)
-
-
-def run_roboswag():
-    generate_cli()
+run_roboswag()

roboswag/cli.py

@@ -0,0 +1,22 @@
+import argparse
+import sys
+
+from roboswag.version import __version__
+from roboswag.generate.generate import generate
+
+
+def generate_cli():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-s", "--spec", metavar="SWAGGER", help="OpenAPI specification file")
+    parser.add_argument(
+        "-v", "--version", action="version", version=__version__, help="display Roboswag version and exit"
+    )
+    args = parser.parse_args()
+    if not args.spec:
+        print("Please provide specification file with '-s' option")
+        sys.exit(1)
+    generate(args.spec)
+
+
+def run_roboswag():
+    generate_cli()

roboswag/generate/generate.py

@@ -101,6 +101,6 @@ def generate_schemas(swagger, output_dir):
         with open(schema_file, "w") as f:
             f.write(json.dumps(schema, indent=4))
             f.write("\n")
-        print(f"Generated '{output_dir}\\{schema_name}.json' file")
+        print(f"Generated '{schema_file}' file")
         schema_files.append(schema_file)
     return schema_files

roboswag/model.py

@@ -0,0 +1,77 @@
+import requests
+import urllib3
+
+from roboswag.auth import TokenHandler
+from roboswag.logger import Logger
+
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+
+class APIModel:
+    def __init__(
+        self,
+        base_url,
+        verify=False,
+        headers=None,
+        content_type="application/json",
+        proxies=None,
+        allow_redirects=True,
+        authorization=TokenHandler,
+    ):
+        # Set headers from init too (or reuse auth)
+        self.base_url = base_url
+        self.session = requests.Session()
+        self.session.verify = verify
+        self.content_type = content_type
+        if content_type is not None:
+            self.session.headers = {"Content-Type": content_type}
+        self.allow_redirects = allow_redirects
+        if proxies is not None:  # TODO urllib have autodetect proxy - allow to use it
+            self.session.proxies.update(proxies)
+        self.authorization = authorization
+        self.logger = Logger()
+        if headers is not None:
+            self.session.headers.update(headers)
+
+    def send_request(self, method, url, status=None, headers=None, body=None, query=None, **kwargs):
+        headers = self.trim_empty(headers)
+        query = self.trim_empty(query)
+        # auth = self.authorization(**kwargs) if self.authorization is not None else None
+        content_type = kwargs.get("content-type", self.content_type)
+        if content_type is not None:
+            headers["Content-Type"] = content_type
+
+        resp = self.session.request(
+            method,
+            url=self.base_url + url,
+            headers=headers,
+            json=body,
+            params=query,
+            # auth=auth,
+            allow_redirects=self.allow_redirects,
+        )
+        # TODO quiet mode
+        self.logger.log_request(resp)
+        self.logger.log_response(resp)
+        if status is not None:
+            assert resp.status_code == status, f"Expected return status: {status} but received: {resp.status_code}"
+        return resp
+
+    def post(self, *args, **kwargs):
+        # TODO handle files upload
+        return self.send_request("POST", *args, **kwargs)
+
+    def get(self, *args, **kwargs):
+        return self.send_request("GET", *args, **kwargs)
+
+    def put(self, *args, **kwargs):
+        return self.send_request("PUT", *args, **kwargs)
+
+    def delete(self, *args, **kwargs):
+        return self.send_request("DELETE", *args, **kwargs)
+
+    @staticmethod
+    def trim_empty(dictionary):
+        if dictionary is None:
+            return {}
+        return {key: value for key, value in dictionary.items() if value is not None}

roboswag/version.py

@@ -0,0 +1,6 @@
+from importlib.metadata import version
+
+try:
+    __version__ = version("roboswag")
+except Exception:  # pragma: no cover
+    pass