Feat: Python request methods.

Signed-off-by: Caroline Russell <[email protected]>

.github/workflows/test.yml

@@ -27,7 +27,7 @@ jobs:
     - name: Set up Python
       uses: actions/setup-python@v5
       with:
-        python-version: "3.11"
+        python-version: "3.12"
     - name: Make sure versions are synced if PR or ref is tag
       if: ${{ github.event_name == 'pull_request' || github.ref_type == 'tag' }}
       shell: bash

atom_tools/cli/commands/convert.py

@@ -11,6 +11,9 @@
 from atom_tools.lib.utils import export_json
 
 
+logger = logging.getLogger(__name__)
+
+
 class ConvertCommand(Command):
     """
     This command handles the conversion of an atom slice to a specified
@@ -87,6 +90,6 @@ def handle(self):
                     logging.warning('No results produced!')
                     sys.exit(1)
                 export_json(result, self.option('output-file'), 4)
-                logging.info(f'OpenAPI document written to {self.option("output-file")}.')
+                logger.info(f'OpenAPI document written to {self.option("output-file")}.')
             case _:
                 raise ValueError(f'Unknown destination format: {self.option("format")}')

atom_tools/lib/converter.py

@@ -24,7 +24,7 @@
 logger = logging.getLogger(__name__)
 regex = OpenAPIRegexCollection()
 exclusions = ['/content-type', '/application/javascript', '/application/json', '/application/text',
-              '/application/xml', '/*', '/*/*', '/allow']
+              '/application/xml', '/*', '/*/*', '/allow', '/GET', '/POST', '/xml', '/cookie']
 
 
 class OpenAPI:
@@ -157,6 +157,21 @@ def populate_endpoints(self, method_map: Dict) -> Dict[str, Any]:
                         paths_object = new_path_item
         return paths_object
 
+    def _add_py_request_methods(self, paths_item_object: Dict) -> Dict:
+        """Add default request methods for flask and django"""
+        if self.usages.custom_attr == 'flask':
+            paths_item_object = merge_path_objects(
+                paths_item_object,
+                {'head': {'responses': {}}, 'options': {'responses': {}}}
+            )
+            pio_ct = len(paths_item_object.keys())
+            if pio_ct == 2 or (pio_ct == 3 and 'x-atom-usages' in paths_item_object):
+                paths_item_object['get'] = {'responses': {}}
+        elif self.usages.custom_attr == 'django':
+            paths_item_object = merge_path_objects(
+                paths_item_object, {'get': {'responses': {}}, 'post': {'responses': {}}})
+        return paths_item_object
+
     def _calls_to_params(self, ep: str, orig_ep: str, call: Dict | None) -> Dict[str, Any]:
         """
         Transforms a call and endpoint into a parameter object and organizes it
@@ -254,6 +269,15 @@ def _extract_params(self, ep: str) -> Tuple[str, bool, List]:
             py_special_case = True
         return ep, py_special_case, tmp_params
 
+    def _extract_unparsed_params(
+            self, ep: str, orig_ep: str, py_special_case: bool, tmp_params: List
+    ) -> Tuple[str, List]:
+        if ':' in ep or '<' in ep:
+            ep, py_special_case, tmp_params = self._extract_params(ep)
+        if '{' in ep and not py_special_case:
+            tmp_params = self._generic_params_helper(ep, orig_ep)
+        return ep, tmp_params
+
     def _filter_matches(self, matches: List[str], code: str) -> List[str]:
         """
         Filters a list of matches based on certain criteria.
@@ -311,8 +335,8 @@ def _identify_target_line_nums(self, methods: Dict[str, Any]) -> Dict:
         file_names = list(methods['file_names'].keys())
         if not file_names:
             return {}
-        conditional = [f'fileName==`{i}`' for i in file_names]
-        conditional = '*[?' + ' || '.join(conditional) + (  # type: ignore
+        file_names = [f'fileName==`{i}`' for i in file_names]
+        conditional = '*[?' + ' || '.join(file_names) + (
             '][].{file_name: fileName, methods: usages[].targetObj[].{resolved_method: '
             'resolvedMethod || callName || code || name, line_number: lineNumber}}')
         pattern = jmespath.compile(conditional)  # type: ignore
@@ -341,22 +365,23 @@ def _paths_object_helper(
         tmp_params: List = []
         py_special_case = False
         orig_ep = ep
-        if ':' in ep or '<' in ep:
-            ep, py_special_case, tmp_params = self._extract_params(ep)
-        if '{' in ep and not py_special_case:
-            tmp_params = self._generic_params_helper(ep, orig_ep)
+        ep, tmp_params = self._extract_unparsed_params(ep, orig_ep, py_special_case, tmp_params)
         if tmp_params:
             paths_item_object['parameters'] = tmp_params
         if calls:
             for call in calls:
-                paths_item_object |= self._calls_to_params(ep, orig_ep, call)
+                paths_item_object = merge_path_objects(
+                    paths_item_object, self._calls_to_params(ep, orig_ep, call)
+                )
         if (call_line_numbers or line_number) and (line_nos := create_ln_entries(
                 filename, list(set(call_line_numbers)), line_number)):
             if 'x-atom-usages' in paths_item_object:
                 paths_item_object['x-atom-usages'] = merge_x_atom(
                     paths_item_object['x-atom-usages'], line_nos)
             else:
                 paths_item_object |= line_nos
+        if self.usages.origin_type in ('py', 'python'):
+            paths_item_object = self._add_py_request_methods(paths_item_object)
         return ep, paths_item_object
 
     def _parse_path_regexes(self, endpoint: str) -> str:
@@ -413,7 +438,13 @@ def _process_methods(self) -> Dict[str, List[str]]:
             '[].resolvedMethod[]}')
 
         user_defined_types = self._process_methods_helper(
-            'userDefinedTypes[].{file_name: name, resolved_methods: fields[].name}')
+            'userDefinedTypes[].{file_name: fileName, resolved_methods: fields[].name}')
+
+        if self.usages.origin_type in ('py', 'python'):
+            user_defined_types = merge_path_objects(
+                user_defined_types,
+                self._process_methods_helper('userDefinedTypes[].{file_name: fileName, '
+                                             'resolved_methods: procedures[].resolvedMethod}'))
 
         for key, value in calls.items():
             if method_map.get(key):
@@ -454,8 +485,10 @@ def _process_methods_helper(self, pattern: str) -> Dict[str, Any]:
         for r in result:
             file_name = r['file_name']
             methods = r['resolved_methods']
-            resolved.setdefault(file_name, {'resolved_methods': []})[
-                'resolved_methods'].extend(methods)
+            if resolved.get(file_name):
+                resolved[file_name]['resolved_methods'].extend(methods)
+            else:
+                resolved[file_name] = {'resolved_methods': methods}
 
         return resolved
 
@@ -539,7 +572,12 @@ def _query_calls_helper(self, file_name: str) -> List[Dict]:
         pattern = (f'objectSlices[?fileName==`{json.dumps(file_name.encode().decode())}`].usages[]'
                    f'.*[?callName][][]')
         compiled_pattern = jmespath.compile(pattern)
-        return compiled_pattern.search(self.usages.content)
+        result = compiled_pattern.search(self.usages.content)
+        pattern = (f'userDefinedTypes[?fileName==`{json.dumps(file_name.encode().decode())}`][].procedures[]')
+        compiled_pattern = jmespath.compile(pattern)
+        result2 = compiled_pattern.search(self.usages.content)
+        result += result2
+        return result
 
 
 def create_ln_entries(filename: str, call_line_numbers: List, line_number: int | None) -> Dict:
@@ -662,7 +700,9 @@ def merge_path_objects(p1: Dict, p2: Dict) -> Dict:
             continue
         for k, v in value.items():
             if p1[key].get(k):
-                if k == 'x-atom-usages':
+                if k == 'resolved_methods':
+                    p1[key][k].extend(v)
+                elif k == 'x-atom-usages':
                     p1[key][k] = merge_x_atom(p1[key][k], v)
                 elif k == 'parameters':
                     p1[key][k] = merge_params(p1[key][k], v)

atom_tools/lib/slices.py

@@ -65,7 +65,7 @@ def import_flat_slice(content: Dict) -> Dict[str, Dict]:
     return create_flattened_dicts(content)
 
 
-def import_slice(filename: str | Path) -> Tuple[Dict, str]:
+def import_slice(filename: str | Path) -> Tuple[Dict, str, str]:
     """
     Import a slice from a JSON file.
 
@@ -85,12 +85,17 @@ def import_slice(filename: str | Path) -> Tuple[Dict, str]:
     """
     content: Dict = {}
     slice_type = ''
+    custom_attr = ''
     if not filename:
         logger.warning('No filename specified.')
-        return content, slice_type
+        return content, slice_type, custom_attr
     try:
         with open(filename, 'r', encoding='utf-8') as f:
             raw_content = f.read().replace(r'\\', '/')
+            if 'flask' in raw_content:
+                custom_attr = 'flask'
+            elif 'django' in raw_content:
+                custom_attr = 'django'
             content = json.loads(raw_content)
         if content.get('objectSlices'):
             slice_type = 'usages'
@@ -107,7 +112,7 @@ def import_slice(filename: str | Path) -> Tuple[Dict, str]:
     if not slice_type:
         logger.warning('Slice type not recognized.')
         sys.exit(1)
-    return content, slice_type
+    return content, slice_type, custom_attr
 
 
 def process_attrib_dict(attrib_dict: Dict, k: str, v: str) -> Dict:
@@ -136,7 +141,7 @@ class AtomSlice:
     """
 
     def __init__(self, filename: str | Path, origin_type: str | None = None) -> None:
-        self.content, self.slice_type = import_slice(filename)
+        self.content, self.slice_type, self.custom_attr = import_slice(filename)
         self.origin_type = origin_type
 
 
@@ -149,5 +154,5 @@ class FlatSlice:
     attrib_dicts: Dict = field(default_factory=dict)
 
     def __post_init__(self):
-        self.content, self.slice_type = import_slice(self.slice_file)
+        self.content, self.slice_type, self.custom_attr = import_slice(self.slice_file)
         self.attrib_dicts = import_flat_slice(self.content)

atom_tools/lib/utils.py

@@ -2,26 +2,16 @@
 import json
 import logging
 import re
-from typing import Dict, Mapping
+from typing import Dict
 
 
 logger = logging.getLogger(__name__)
 
 
-def merge_dicts(d1: Dict, d2: Dict) -> Dict:
-    """Merges two dictionaries"""
-    for k in d2:
-        if k in d1 and isinstance(d1[k], dict) and isinstance(d2[k], Mapping):
-            merge_dicts(d1[k], d2[k])
-        else:
-            d1[k] = d2[k]
-    return d1
-
-
 def export_json(data: Dict, outfile: str, indent: int | None = None) -> None:
     """Exports data to json"""
     with open(outfile, 'w', encoding='utf-8') as f:
-        json.dump(data, f, indent=indent)
+        json.dump(data, f, indent=indent, sort_keys=True)
 
 
 def add_outfile_to_cmd(cmd: str, outfile: str):