Tjm/access

example/resources/form.py

@@ -5,6 +5,7 @@
 """
 
 from reliqua.resources.base import Resource
+from reliqua.status_codes import HTTP
 
 
 class Contact(Resource):
@@ -34,3 +35,6 @@ def on_post(self, req, resp):
         p = req.params
         if p.get("subject"):
             resp.media = {"success": True}
+        else:
+            resp.status = HTTP("400")
+            resp.media = {"error": "Subject is required"}

example/resources/media.py

@@ -7,6 +7,7 @@
 import os
 
 from reliqua.resources.base import Resource
+from reliqua.status_codes import HTTP
 
 
 class Gzip(Resource):
@@ -16,19 +17,25 @@ class Gzip(Resource):
         "/gzip": {},
     }
 
-    def on_get(self, req, resp):
+    def on_get(self, _req, resp):
         """
         Send contact message.
 
         :accepts json:          Accepts JSON
-        :response 200 binary:       All good
-        :return gzip:      Return data
+        :response 200 binary:   All good
+        :return gzip:           Return data
         """
-        fh = open("/tmp/hello.txt.gz", "rb")
-        resp.append_header("Content-Disposition", "attachment; filename=hello.txt.gz")
-        resp.content_type = "application/gzip"
-        resp.content_encoding = "gzip"
-        resp.data = fh.read()
+        try:
+            with open("/tmp/hello.txt.gz", "rb") as fh:
+                resp.append_header("Content-Disposition", "attachment; filename=hello.txt.gz")
+                resp.content_type = "application/gzip"
+                resp.content_encoding = "gzip"
+                resp.data = fh.read()
+        except FileNotFoundError:
+            resp.status = HTTP("404")
+        except Exception as e:
+            resp.status = HTTP("500")
+            resp.media = {"error": str(e)}
 
 
 class Binary(Resource):
@@ -38,16 +45,23 @@ class Binary(Resource):
         "/bin/{filename}": {},
     }
 
-    def on_get(self, req, resp, filename):
+    def on_get(self, _req, resp, filename):
         """
         Send contact message.
 
         :param str filename:    [in=path] Filename
-        :response 200 binary:       All good
+        :response 200 binary:   All good
         :return binary:         Return data
         """
         path = f"/tmp/{filename}"
-        resp.stream = open(path, "rb")
-        resp.content_type = "application/gzip"
-        resp.content_encoding = "gzip"
-        resp.content_length = os.path.getsize(path)
+        try:
+            with open(path, "rb") as fh:
+                resp.stream = fh
+                resp.content_type = "application/gzip"
+                resp.content_encoding = "gzip"
+                resp.content_length = os.path.getsize(path)
+        except FileNotFoundError:
+            resp.status = HTTP("404")
+        except Exception as e:
+            resp.status = HTTP("500")
+            resp.media = {"error": str(e)}

example/resources/servers.py

@@ -26,20 +26,20 @@ class Server(Resource):
 
     def on_get_by_id(self, _req, resp, id=None):
         """
-        Retrieve servers.
+        Retrieve a server.
 
-        Retrieve a list of servers in the lab.
+        Retrieve a server by its ID.
 
         :param str id:       [in=path, required] Server ID
 
-        :response 200:       server was retrieved
-        :response 400:       invalid query parameter
+        :response 200:       Server was retrieved
+        :response 404:       Server not found
 
         :return json:
         """
         try:
-            resp.media = servers[id]
-        except IndexError:
+            resp.media = servers[int(id)]
+        except (IndexError, ValueError):
             resp.status = HTTP("404")
 
 
@@ -54,14 +54,14 @@ class Servers(Resource):
 
     def on_get(self, req, resp):
         """
-        Retrieve a server.
+        Retrieve servers.
 
-        Retrieve server information
+        Retrieve a list of servers in the lab.
 
         :param list labs:     [in=query] The labs servers are located
 
-        :response 200:       server  was retrieved
-        :response 400:       invalid query parameter
+        :response 200:        Servers were retrieved
+        :response 400:        Invalid query parameter
 
         :return json:
         """
@@ -70,18 +70,18 @@ def on_get(self, req, resp):
 
     def on_get_by_cpu(self, _req, resp, cpus=1):
         """
-        Retrieve a server by cpu.
+        Retrieve a server by CPU.
 
-        Retrieve server information by cpu
+        Retrieve server information by CPU.
 
         :param int cpus:     [in=path required] Number of CPUs for server
 
-        :response 200:       server  was retrieved
-        :response 400:       invalid query parameter
+        :response 200:       Server was retrieved
+        :response 404:       Server not found
 
         :return json:
         """
         try:
-            resp.media = servers[cpus]
-        except IndexError:
+            resp.media = servers[int(cpus)]
+        except (IndexError, ValueError):
             resp.status = HTTP("404")

example/resources/users.py

@@ -64,7 +64,7 @@ def on_get_by_id(self, _req, resp, id=None):
         """
         try:
             resp.media = users[int(id)]
-        except IndexError:
+        except (IndexError, ValueError):
             resp.status = HTTP("404")
 
     def on_delete_by_id(self, _req, resp, id=None):
@@ -78,7 +78,7 @@ def on_delete_by_id(self, _req, resp, id=None):
         try:
             users.pop(int(id))
             resp.media = {"success": True}
-        except IndexError:
+        except (IndexError, ValueError):
             resp.status = HTTP("400")
 
 

src/reliqua/api.py

@@ -65,7 +65,7 @@ def __init__(
         :param dict info:                     Application information
         :param dict openapi:                  OpenAPI configuration options
 
-        :return:                               api instance
+        :return:                               API instance
         """
         info = info or {}
         openapi = openapi or {}
@@ -107,6 +107,7 @@ def __init__(
         self._add_docs()
 
     def _add_handlers(self):
+        """Add custom media handlers."""
         extra_handlers = {
             "application/yaml": YAMLHandler(),
             "text/html; charset=utf-8": TextHandler(),
@@ -118,6 +119,7 @@ def _add_handlers(self):
         self.resp_options.media_handlers.update(extra_handlers)
 
     def _load_resources(self):
+        """Load resource classes from the specified resource path."""
         resources = []
         path = f"{self.resource_path}/*.py"
         print(f"searching {path}")
@@ -128,10 +130,11 @@ def _load_resources(self):
             classes = self._get_classes(file)
             resources.extend(classes)
 
-        # add config to resource
+        # Add config to resource
         self.resources = [x(app_config=self.config, **self.resource_attributes) for x in resources]
 
     def _is_route_method(self, name, suffix):
+        """Check if a method name is a route method."""
         if not name.startswith("on_"):
             return None
 
@@ -141,6 +144,7 @@ def _is_route_method(self, name, suffix):
         return re.search(r"^on_([a-z]+)$", name)
 
     def _parse_methods(self, resource, route, methods):
+        """Parse methods of a resource for a given route."""
         parser = SphinxParser()
         for name in methods:
             operation_id = f"{resource.__class__.__name__}.{name}"
@@ -149,23 +153,30 @@ def _parse_methods(self, resource, route, methods):
             resource.__data__[route][action] = parser.parse(method, operation_id=operation_id)
 
     def _parse_resource(self, resource):
+        """Parse a resource to extract routes and methods."""
         for route, data in resource.__routes__.items():
             resource.__data__[route] = {}
             suffix = data.get("suffix", None)
             methods = [x for x in dir(resource) if self._is_route_method(x, suffix)]
             self._parse_methods(resource, route, methods)
 
     def _parse_docstrings(self):
+        """Parse docstrings of all resources."""
         for resource in self.resources:
             resource.__data__ = {}
             self._parse_resource(resource)
 
     def _get_classes(self, filename):
+        """Get resource classes from a file."""
         classes = []
         module_name = str(uuid.uuid3(uuid.NAMESPACE_OID, filename))
         spec = importlib.util.spec_from_file_location(module_name, filename)
         module = importlib.util.module_from_spec(spec)
-        spec.loader.exec_module(module)
+        try:
+            spec.loader.exec_module(module)
+        except (ImportError, FileNotFoundError, SyntaxError, TypeError, AttributeError) as e:
+            print(f"Error loading module {module_name}: {e}")
+            return classes
 
         for _, c in inspect.getmembers(module, inspect.isclass):
             if issubclass(c, Resource) and hasattr(c, "__routes__"):
@@ -174,12 +185,14 @@ def _get_classes(self, filename):
         return classes
 
     def _add_routes(self):
+        """Add routes for all resources."""
         for resource in self.resources:
             routes = resource.__routes__
             for route, kwargs in routes.items():
                 self.add_route(route, resource, **kwargs)
 
     def _add_docs(self):
+        """Add Swagger and OpenAPI documentation routes."""
         swagger = Swagger(
             self.openapi["static_url"],
             self.openapi["spec_url"],

src/reliqua/app.py

@@ -43,18 +43,19 @@
 
 def update_dict(a, b):
     """
-    Update dictionary A with values from dictionary B.
+    Update dictionary a with values from dictionary b.
 
-    Update the diction A with values from dictionary B. Remove
-    keys from A that are not in B.
+    Update dictionary a with values from dictionary b. Remove
+    keys from a that are not in b.
 
-    :param dict a:    Dictionary A
-    :param dict b:    Dictionary B
+    :param dict a:  Dictionary to update
+    :param dict b:  Dictionary with new values
+    :return dict:   Updated dictionary
     """
-    # Create a new dictionary with keys from A that are also in B
+    # Create a new dictionary with keys from a that are also in b
     updated_dict = {key: a[key] for key in a if key in b}
 
-    # Add keys from B that are not in A
+    # Add keys from b that are not in a
     for key in b:
         if key not in updated_dict:
             updated_dict[key] = b[key]
@@ -78,8 +79,9 @@ def load_config(config_file):
 
         for option in config.options(section):
             params[option] = config.get(section, option)
-    except TypeError:
-        pass
+    except (TypeError, configparser.Error) as e:
+        # Log the error or handle it appropriately
+        print(f"Error loading config file: {e}")
 
     return params
 
@@ -125,7 +127,7 @@ def __init__(
 
         middleware.append(ProcessParams())
 
-        # trim slashes from proxy URL if specified; otherwise set default proxy url
+        # Trim slashes from proxy URL if specified; otherwise set default proxy URL
         bind = self.gunicorn_options["bind"]
         openapi["api_url"] = openapi["api_url"].rstrip("/") if openapi["api_url"] else f"http://{bind}"
 

src/reliqua/docs.py

@@ -58,14 +58,16 @@ def __init__(self, schema):
         :param dict schema:    Documents JSON schema
         :return:               None
         """
+        super().__init__()
         self.schema = schema
 
     def on_get(self, _req, resp):
         """
         Return the JSON document schema.
 
-        :param Response response:    Response object
-        :return:                     None
+        :param Request _req:       Request object
+        :param Response resp:      Response object
+        :return:                   None
         """
         resp.set_header("Access-Control-Allow-Origin", "*")
         resp.media = self.schema

src/reliqua/openapi.py

@@ -115,7 +115,7 @@ def __init__(
         self.examples = examples
         self._datatype = datatype
 
-        # path parameters are always required
+        # Path parameters are always required
         if self.location == "path":
             self.required = True
 
@@ -125,7 +125,7 @@ def datatype(self):
         value = self._datatype
         a, _, b = value.partition("|")
 
-        # handle defined list type ex: list[str]
+        # Handle defined list type ex: list[str]
         if re.search(r"list\[(\w+)]", a):
             a = "list"