From d8a3a9cd9fd9db8d9b6f4ea6aa4ce261fe9117c2 Mon Sep 17 00:00:00 2001 From: "John M. Kuchta" Date: Tue, 16 Jan 2024 10:51:11 -0800 Subject: [PATCH 1/3] Draft revision to resolve #241 and other similar potential issues. Includes many black formatting updates. --- generator/generate_library.py | 635 ++++++++++++++++++++-------------- 1 file changed, 381 insertions(+), 254 deletions(-) diff --git a/generator/generate_library.py b/generator/generate_library.py index 7c035d30..ffda0f9b 100644 --- a/generator/generate_library.py +++ b/generator/generate_library.py @@ -20,56 +20,61 @@ API key can, and is recommended to, be set as an environment variable named MERAKI_DASHBOARD_API_KEY. """ -REVERSE_PAGINATION = ['getNetworkEvents', 'getOrganizationConfigurationChanges'] +REVERSE_PAGINATION = ["getNetworkEvents", "getOrganizationConfigurationChanges"] # Helper function to return pagination parameters depending on endpoint def generate_pagination_parameters(operation: str): ret = { - 'total_pages': { - 'type': 'integer or string', - 'description': 'use with perPage to get total results up to total_pages*perPage; -1 or "all" for all pages', + "total_pages": { + "type": "integer or string", + "description": 'use with perPage to get total results up to total_pages*perPage; -1 or "all" for all pages', }, - 'direction': { - 'type': 'string', - 'description': 'direction to paginate, either "next" or "prev" (default) page' + "direction": { + "type": "string", + "description": 'direction to paginate, either "next" or "prev" (default) page' if operation in REVERSE_PAGINATION else 'direction to paginate, either "next" (default) or "prev" page', - } + }, } - if operation == 'getNetworkEvents': - ret['event_log_end_time'] = {'type': 'string', - 'description': 'ISO8601 Zulu/UTC time, to use in conjunction with startingAfter, ' - 'to retrieve events within a time window'} + if operation == "getNetworkEvents": + ret["event_log_end_time"] = { + "type": "string", + "description": "ISO8601 Zulu/UTC time, to use in conjunction with startingAfter, " + "to retrieve events within a time window", + } return ret def check_python_version(): # Check minimum Python version version_warning_string = ( - f"This library generator requires Python 3.10 at minimum. " - f"Your interpreter version is: {platform.python_version()}. " - f"Please consult the readme at your convenience: " - f"https://github.com/meraki/dashboard-api-python/blob/main/generator/readme.md " - f"Additional details: " - f"python_version_tuple()[0] = {platform.python_version_tuple()[0]}; " - f"python_version_tuple()[1] = {platform.python_version_tuple()[1]} " - ) - - if not (int(platform.python_version_tuple()[0]) == 3 and int(platform.python_version_tuple()[1]) >= 10): + f"This library generator requires Python 3.10 at minimum. " + f"Your interpreter version is: {platform.python_version()}. " + f"Please consult the readme at your convenience: " + f"https://github.com/meraki/dashboard-api-python/blob/main/generator/readme.md " + f"Additional details: " + f"python_version_tuple()[0] = {platform.python_version_tuple()[0]}; " + f"python_version_tuple()[1] = {platform.python_version_tuple()[1]} " + ) + + if not ( + int(platform.python_version_tuple()[0]) == 3 + and int(platform.python_version_tuple()[1]) >= 10 + ): sys.exit(version_warning_string) # Returns full link to endpoint's documentation on Developer Hub # Note: updates to the documentation site may impact these URLs. def docs_url(operation: str): - base_url = 'https://developer.cisco.com/meraki/api-v1/#!' - ret = '' + base_url = "https://developer.cisco.com/meraki/api-v1/#!" + ret = "" for letter in operation: if letter.islower(): ret += letter else: - ret += f'-{letter.lower()}' + ret += f"-{letter.lower()}" return base_url + ret @@ -80,49 +85,68 @@ def return_params(operation: str, params: dict, param_filters): return params else: ret = {} - if 'required' in param_filters: - ret.update({k: v for k, v in params.items() if 'required' in v and v['required']}) - if 'pagination' in param_filters: - ret.update(generate_pagination_parameters(operation) if 'perPage' in params else {}) - if 'optional' in param_filters: - ret.update({k: v for k, v in params.items() if 'required' in v and not v['required']}) - if 'path' in param_filters: - ret.update({k: v for k, v in params.items() if 'in' in v and v['in'] == 'path'}) - if 'query' in param_filters: - ret.update({k: v for k, v in params.items() if 'in' in v and v['in'] == 'query'}) - if 'body' in param_filters: - ret.update({k: v for k, v in params.items() if 'in' in v and v['in'] == 'body'}) - if 'array' in param_filters: - ret.update({k: v for k, v in params.items() if 'in' in v and v['type'] == 'array'}) - if 'enum' in param_filters: - ret.update({k: v for k, v in params.items() if 'enum' in v}) + if "required" in param_filters: + ret.update( + {k: v for k, v in params.items() if "required" in v and v["required"]} + ) + if "pagination" in param_filters: + ret.update( + generate_pagination_parameters(operation) if "perPage" in params else {} + ) + if "optional" in param_filters: + ret.update( + { + k: v + for k, v in params.items() + if "required" in v and not v["required"] + } + ) + if "path" in param_filters: + ret.update( + {k: v for k, v in params.items() if "in" in v and v["in"] == "path"} + ) + if "query" in param_filters: + ret.update( + {k: v for k, v in params.items() if "in" in v and v["in"] == "query"} + ) + if "body" in param_filters: + ret.update( + {k: v for k, v in params.items() if "in" in v and v["in"] == "body"} + ) + if "array" in param_filters: + ret.update( + {k: v for k, v in params.items() if "in" in v and v["type"] == "array"} + ) + if "enum" in param_filters: + ret.update({k: v for k, v in params.items() if "enum" in v}) return ret -def unpack_param_without_schema(all_params: dict, this_param: dict, name: str, is_required: bool): - +def unpack_param_without_schema( + all_params: dict, this_param: dict, name: str, is_required: bool +): # Set required attribute - all_params[name] = {'required': is_required} + all_params[name] = {"required": is_required} # Assign relevant attributes - for attribute in ('in', 'type'): + for attribute in ("in", "type"): all_params[name][attribute] = this_param[attribute] # Capture the enum if available - if 'enum' in this_param: - all_params[name]['enum'] = this_param['enum'] + if "enum" in this_param: + all_params[name]["enum"] = this_param["enum"] # Assign the description to the parameter if it's available - if 'description' in this_param: - all_params[name]['description'] = this_param['description'] + if "description" in this_param: + all_params[name]["description"] = this_param["description"] # Fall back to required if there is no description elif is_required: - all_params[name]['description'] = '(required)' + all_params[name]["description"] = "(required)" # Fall back to whatever the description is otherwise else: - all_params[name]['description'] = this_param['description'] + all_params[name]["description"] = this_param["description"] return all_params @@ -131,30 +155,30 @@ def unpack_param_with_schema(all_params: dict, this_param: dict): # the parameter will have a top-level object 'schema' and within that, 'properties' in OASv2 # in OASv3, the parameter will only have this for query and path parameters, and requestBody params # will be in a separate key - keys = this_param['schema']['properties'] + keys = this_param["schema"]["properties"] # parse the properties and assign types and descriptions for k in keys: # if required, set required true - if 'required' in this_param['schema'] and k in this_param['schema']['required']: - all_params[k] = {'required': True} + if "required" in this_param["schema"] and k in this_param["schema"]["required"]: + all_params[k] = {"required": True} else: - all_params[k] = {'required': False} + all_params[k] = {"required": False} # identify whether the parameter is in the path or query, or for OASv2, in the body - all_params[k]['in'] = this_param['in'] + all_params[k]["in"] = this_param["in"] # assign the right data type/description to the parameter per the schema - for attribute in ('type', 'description'): + for attribute in ("type", "description"): all_params[k][attribute] = keys[k][attribute] # capture schema enum if available - if 'enum' in keys[k]: - all_params[k]['enum'] = keys[k]['enum'] + if "enum" in keys[k]: + all_params[k]["enum"] = keys[k]["enum"] # capture schema example if available - if 'example' in this_param['schema'] and k in this_param['schema']['example']: - all_params[k]['example'] = this_param['schema']['example'][k] + if "example" in this_param["schema"] and k in this_param["schema"]["example"]: + all_params[k]["example"] = this_param["schema"]["example"][k] return all_params @@ -166,22 +190,26 @@ def unpack_params(operation: str, parameters: dict, param_filters): # Iterate through the endpoint's parameters for p in parameters: # Name the parameter - name = p['name'] + name = p["name"] # Consult the schema if there is one - if 'schema' in p: + if "schema" in p: unpacked_params.update(unpack_param_with_schema(unpacked_params, p)) # If there is no schema, then consult the required attribute if it exists - elif 'required' in p and p['required']: - unpacked_params.update(unpack_param_without_schema(unpacked_params, p, name, True)) + elif "required" in p and p["required"]: + unpacked_params.update( + unpack_param_without_schema(unpacked_params, p, name, True) + ) # Otherwise the parameter is not required else: - unpacked_params.update(unpack_param_without_schema(unpacked_params, p, name, False)) + unpacked_params.update( + unpack_param_without_schema(unpacked_params, p, name, False) + ) # Add custom library parameters to handle pagination - if 'perPage' in unpacked_params: + if "perPage" in unpacked_params: unpacked_params.update(generate_pagination_parameters(operation)) # Return parameters based on matching input filters @@ -200,41 +228,74 @@ def parse_params(operation: str, parameters: dict, param_filters=None): def generate_library(spec: dict, version_number: str, is_github_action: bool): # Supported scopes list will include organizations, networks, devices, and all product types. - supported_scopes = ['organizations', 'networks', 'devices', 'appliance', 'camera', 'cellularGateway', 'insight', - 'sm', 'switch', 'wireless', 'sensor', 'administered', 'licensing', 'secureConnect'] + supported_scopes = [ + "organizations", + "networks", + "devices", + "appliance", + "camera", + "cellularGateway", + "insight", + "sm", + "switch", + "wireless", + "sensor", + "administered", + "licensing", + "secureConnect", + ] # legacy scopes = ['organizations', 'networks', 'devices', 'appliance', 'camera', 'cellularGateway', 'insight', # 'sm', 'switch', 'wireless'] - tags = spec['tags'] - paths = spec['paths'] + tags = spec["tags"] + paths = spec["paths"] # Scopes used when generating the library will depend on the provided version of the API spec. - scopes = {tag['name']: {} for tag in tags if tag['name'] in supported_scopes} + scopes = {tag["name"]: {} for tag in tags if tag["name"] in supported_scopes} - batchable_action_summaries = [action['summary'] for action in spec['x-batchable-actions']] + batchable_actions = spec["x-batchable-actions"] # Set template_dir if a GitHub action is invoking it if is_github_action: - template_dir = 'generator/' + template_dir = "generator/" else: - template_dir = '' + template_dir = "" # Check paths and create directories if needed - directories = ['meraki', 'meraki/api', 'meraki/api/batch', 'meraki/aio', 'meraki/aio/api', 'meraki/api/batch'] + directories = [ + "meraki", + "meraki/api", + "meraki/api/batch", + "meraki/aio", + "meraki/aio/api", + "meraki/api/batch", + ] for directory in directories: if not os.path.isdir(directory): os.mkdir(directory) # Files that are not generated - non_generated = ['__init__.py', 'config.py', 'common.py', 'exceptions.py', 'rest_session.py', 'api/__init__.py', - 'aio/__init__.py', 'aio/rest_session.py', 'aio/api/__init__.py', 'api/batch/__init__.py'] - base_url = 'https://raw.githubusercontent.com/meraki/dashboard-api-python/master/meraki/' + non_generated = [ + "__init__.py", + "config.py", + "common.py", + "exceptions.py", + "rest_session.py", + "api/__init__.py", + "aio/__init__.py", + "aio/rest_session.py", + "aio/api/__init__.py", + "api/batch/__init__.py", + ] + base_url = ( + "https://raw.githubusercontent.com/meraki/dashboard-api-python/master/meraki/" + ) for file in non_generated: - response = requests.get(f'{base_url}{file}') - with open(f'meraki/{file}', 'w+', encoding='utf-8', newline=None) as fp: + response = requests.get(f"{base_url}{file}") + with open(f"meraki/{file}", "w+", encoding="utf-8", newline=None) as fp: contents = response.text - if file == '__init__.py': - start = contents.find('__version__ = ') - end = contents.find('\n', start) - contents = f'{contents[:start]}__version__ = \'{version_number}\'{contents[end:]}' + if file == "__init__.py": + start = contents.find("__version__ = ") + end = contents.find("\n", start) + contents = f"{contents[:start]}__version__ = '{version_number}'{contents[end:]}" fp.write(contents) # Organize data from OpenAPI specification @@ -242,15 +303,14 @@ def generate_library(spec: dict, version_number: str, is_github_action: bool): for path, methods in paths.items(): # method is the HTTP action, e.g. get, put, etc. for method in methods: - # endpoint is the method for that specific path endpoint = paths[path][method] # the endpoint has tags - tags = endpoint['tags'] + tags = endpoint["tags"] # the endpoint has an operationId - operation = endpoint['operationId'] + operation = endpoint["operationId"] # add the operation ID to the list operations.append(operation) @@ -266,137 +326,174 @@ def generate_library(spec: dict, version_number: str, is_github_action: bool): scopes[scope][path][method] = endpoint # Inform the user of the number of operations found - print(f'Total of {len(operations)} endpoints found from OpenAPI spec...') + print(f"Total of {len(operations)} endpoints found from OpenAPI spec...") # Generate API libraries # We will use newline=None to ensure that line breaks are handled correctly, especially when generating # on Windows and using `git autocrlf true` - jinja_env = jinja2.Environment(trim_blocks=True, lstrip_blocks=True, keep_trailing_newline=True) + jinja_env = jinja2.Environment( + trim_blocks=True, lstrip_blocks=True, keep_trailing_newline=True + ) # Iterate through the scopes creating standard, asyncio and batch modules for each - generate_modules(batchable_action_summaries, jinja_env, scopes, template_dir) + generate_modules(batchable_actions, jinja_env, scopes, template_dir) -def generate_modules(batchable_action_summaries, jinja_env, scopes, template_dir): +def generate_modules(batchable_actions, jinja_env, scopes, template_dir): for scope in scopes: - print(f'...generating {scope}') + print(f"...generating {scope}") section = scopes[scope] # Generate the standard module - with open(f'meraki/api/{scope}.py', 'w', encoding='utf-8', newline=None) as output: - + with open( + f"meraki/api/{scope}.py", "w", encoding="utf-8", newline=None + ) as output: # Open module file for Asyncio API libraries - async_output = open(f'meraki/aio/api/{scope}.py', 'w', encoding='utf-8', newline=None) + async_output = open( + f"meraki/aio/api/{scope}.py", "w", encoding="utf-8", newline=None + ) # Open module file for Action Batch API libraries - batch_output = open(f'meraki/api/batch/{scope}.py', 'w', encoding='utf-8', newline=None) + batch_output = open( + f"meraki/api/batch/{scope}.py", "w", encoding="utf-8", newline=None + ) modules = [ + {"template_name": "class_template.jinja2", "module_output": output}, { - 'template_name': 'class_template.jinja2', - 'module_output': output + "template_name": "async_class_template.jinja2", + "module_output": async_output, }, { - 'template_name': 'async_class_template.jinja2', - 'module_output': async_output + "template_name": "batch_class_template.jinja2", + "module_output": batch_output, }, - { - 'template_name': 'batch_class_template.jinja2', - 'module_output': batch_output - } ] # Generate modules for module in modules: - render_class_template(jinja_env, template_dir, module['template_name'], module['module_output'], scope) + render_class_template( + jinja_env, + template_dir, + module["template_name"], + module["module_output"], + scope, + ) # Generate API & Asyncio API functions - generate_standard_and_async_functions(jinja_env, template_dir, section, output, async_output) + generate_standard_and_async_functions( + jinja_env, template_dir, section, output, async_output + ) # Generate API action batch functions - generate_action_batch_functions(jinja_env, template_dir, section, batch_output, batchable_action_summaries) + generate_action_batch_functions( + jinja_env, + template_dir, + section, + batch_output, + batchable_actions, + ) -def generate_standard_and_async_functions(jinja_env: jinja2.Environment, template_dir: str, section: dict, - output: open, async_output: open): +def generate_standard_and_async_functions( + jinja_env: jinja2.Environment, + template_dir: str, + section: dict, + output: open, + async_output: open, +): for path, methods in section.items(): for method, endpoint in methods.items(): # Get metadata - tags = endpoint['tags'] - operation = endpoint['operationId'] - description = endpoint['summary'] + tags = endpoint["tags"] + operation = endpoint["operationId"] + description = endpoint["summary"] # will need updating for OASv3 - parameters = endpoint['parameters'] if 'parameters' in endpoint else None + parameters = endpoint["parameters"] if "parameters" in endpoint else None # Function definition - definition = '' + definition = "" if parameters: - for p, values in parse_params(operation, parameters, 'required').items(): - if values['type'] == 'array': - definition += f', {p}: list' - elif values['type'] == 'number': - definition += f', {p}: float' - elif values['type'] == 'integer': - definition += f', {p}: int' - elif values['type'] == 'boolean': - definition += f', {p}: bool' - elif values['type'] == 'object': - definition += f', {p}: dict' - elif values['type'] == 'string': - definition += f', {p}: str' - - if 'perPage' in parse_params(operation, parameters): + for p, values in parse_params( + operation, parameters, "required" + ).items(): + if values["type"] == "array": + definition += f", {p}: list" + elif values["type"] == "number": + definition += f", {p}: float" + elif values["type"] == "integer": + definition += f", {p}: int" + elif values["type"] == "boolean": + definition += f", {p}: bool" + elif values["type"] == "object": + definition += f", {p}: dict" + elif values["type"] == "string": + definition += f", {p}: str" + + if "perPage" in parse_params(operation, parameters): if operation in REVERSE_PAGINATION: definition += ", total_pages=1, direction='prev'" else: definition += ", total_pages=1, direction='next'" - if operation == 'getNetworkEvents': - definition += ', event_log_end_time=None' + if operation == "getNetworkEvents": + definition += ", event_log_end_time=None" - if parse_params(operation, parameters, ['optional']): - definition += ', **kwargs' + if parse_params(operation, parameters, ["optional"]): + definition += ", **kwargs" # Docstring param_descriptions = list() - all_params = parse_params(operation, parameters, ['required', 'pagination', 'optional']) + all_params = parse_params( + operation, parameters, ["required", "pagination", "optional"] + ) if all_params: for p, values in all_params.items(): - param_descriptions.append(f'{p} ({values["type"]}): {values["description"]}') + param_descriptions.append( + f'{p} ({values["type"]}): {values["description"]}' + ) # Combine keyword args with locals - kwarg_line = '' - if parse_params(operation, parameters, ['optional']): - kwarg_line = 'kwargs.update(locals())' - elif parse_params(operation, parameters, ['query', 'array', 'body']): - kwarg_line = 'kwargs = locals()' + kwarg_line = "" + if parse_params(operation, parameters, ["optional"]): + kwarg_line = "kwargs.update(locals())" + elif parse_params(operation, parameters, ["query", "array", "body"]): + kwarg_line = "kwargs = locals()" # Assert valid values for enum - enum_params = parse_params(operation, parameters, ['enum']) + enum_params = parse_params(operation, parameters, ["enum"]) assert_blocks = list() if enum_params: for p, values in enum_params.items(): - assert_blocks.append((p, values['enum'])) + assert_blocks.append((p, values["enum"])) # Function body for GET endpoints query_params = array_params = body_params = path_params = {} - if method == 'get': - array_params, call_line, path_params, query_params = parse_get_params(operation, parameters) + if method == "get": + array_params, call_line, path_params, query_params = parse_get_params( + operation, parameters + ) # Function body for POST/PUT endpoints - elif method == 'post' or method == 'put': - body_params, call_line, path_params = parse_post_and_put_params(method, operation, parameters) + elif method == "post" or method == "put": + body_params, call_line, path_params = parse_post_and_put_params( + method, operation, parameters + ) # Function body for DELETE endpoints - elif method == 'delete': + elif method == "delete": call_line, path_params = parse_delete_params(operation, parameters) # Add function to files - with open(f'{template_dir}function_template.jinja2', encoding='utf-8', newline=None) as fp: + with open( + f"{template_dir}function_template.jinja2", + encoding="utf-8", + newline=None, + ) as fp: function_template = fp.read() template = jinja_env.from_string(function_template) output.write( - '\n\n' + - template.render( + "\n\n" + + template.render( operation=operation, function_definition=definition, description=description, @@ -411,12 +508,12 @@ def generate_standard_and_async_functions(jinja_env: jinja2.Environment, templat array_params=array_params, body_params=body_params, path_params=path_params, - call_line=call_line + call_line=call_line, ) ) async_output.write( - '\n\n' + - template.render( + "\n\n" + + template.render( operation=operation, function_definition=definition, description=description, @@ -431,143 +528,164 @@ def generate_standard_and_async_functions(jinja_env: jinja2.Environment, templat array_params=array_params, body_params=body_params, path_params=path_params, - call_line=call_line + call_line=call_line, ) ) def parse_get_params(operation: str, parameters: dict): - query_params = parse_params(operation, parameters, 'query') - array_params = parse_params(operation, parameters, 'array') - path_params = parse_params(operation, parameters, 'path') - pagination_params = parse_params(operation, parameters, 'pagination') + query_params = parse_params(operation, parameters, "query") + array_params = parse_params(operation, parameters, "array") + path_params = parse_params(operation, parameters, "path") + pagination_params = parse_params(operation, parameters, "pagination") if query_params or array_params: if pagination_params: - if operation == 'getNetworkEvents': - call_line = 'return self._session.get_pages(metadata, resource, params, ' \ - 'total_pages, direction, event_log_end_time)' + if operation == "getNetworkEvents": + call_line = ( + "return self._session.get_pages(metadata, resource, params, " + "total_pages, direction, event_log_end_time)" + ) else: - call_line = 'return self._session.get_pages(metadata, resource, params, ' \ - 'total_pages, direction)' + call_line = ( + "return self._session.get_pages(metadata, resource, params, " + "total_pages, direction)" + ) else: - call_line = 'return self._session.get(metadata, resource, params)' + call_line = "return self._session.get(metadata, resource, params)" else: - call_line = 'return self._session.get(metadata, resource)' + call_line = "return self._session.get(metadata, resource)" return array_params, call_line, path_params, query_params def parse_post_and_put_params(method: str, operation: str, parameters: dict): - body_params = parse_params(operation, parameters, 'body') - path_params = parse_params(operation, parameters, 'path') + body_params = parse_params(operation, parameters, "body") + path_params = parse_params(operation, parameters, "path") if body_params: - call_line = f'return self._session.{method}(metadata, resource, payload)' + call_line = f"return self._session.{method}(metadata, resource, payload)" else: - call_line = f'return self._session.{method}(metadata, resource)' + call_line = f"return self._session.{method}(metadata, resource)" return body_params, call_line, path_params def parse_delete_params(operation: str, parameters: dict): - path_params = parse_params(operation, parameters, 'path') - call_line = 'return self._session.delete(metadata, resource)' + path_params = parse_params(operation, parameters, "path") + call_line = "return self._session.delete(metadata, resource)" return call_line, path_params -def generate_action_batch_functions(jinja_env: jinja2.Environment, template_dir: str, section: dict, - batch_output: open, batchable_action_summaries: list): - +def generate_action_batch_functions( + jinja_env: jinja2.Environment, + template_dir: str, + section: dict, + batch_output: open, + batchable_actions: list, +): for path, methods in section.items(): for method, endpoint in methods.items(): - if endpoint['description'] in batchable_action_summaries: + batchable_action_summaries = [ + action["summary"] for action in batchable_actions + ] + if endpoint["description"] in batchable_action_summaries: # Get metadata - tags = endpoint['tags'] - operation = endpoint['operationId'] - description = endpoint['summary'] + tags = endpoint["tags"] + operation = endpoint["operationId"] + description = endpoint["summary"] + + this_action = [ + action + for action in batchable_actions + if action["summary"] == description + ][0] + + batch_operation = this_action["operation"] + # Function body for POST/PUT endpoints + if method == "post" or method == "put": + # will need update for OASv3 + body_params = parse_params(operation, parameters, "body") + + # Function body for DELETE endpoints is empty (HTTP 204) # May need update for OASv3 - parameters = endpoint['parameters'] if 'parameters' in endpoint else None + parameters = ( + endpoint["parameters"] if "parameters" in endpoint else None + ) # Function definition - definition = '' + definition = "" if parameters: - for p, values in parse_params(operation, parameters, 'required').items(): - + for p, values in parse_params( + operation, parameters, "required" + ).items(): # Match OAS schema types to Python types - match values['type']: - case 'array': - definition += f', {p}: list' - case 'number': - definition += f', {p}: float' - case 'integer': - definition += f', {p}: int' - case 'boolean': - definition += f', {p}: bool' - case 'object': - definition += f', {p}: dict' - case 'string': - definition += f', {p}: str' - - if 'perPage' in parse_params(operation, parameters): + match values["type"]: + case "array": + definition += f", {p}: list" + case "number": + definition += f", {p}: float" + case "integer": + definition += f", {p}: int" + case "boolean": + definition += f", {p}: bool" + case "object": + definition += f", {p}: dict" + case "string": + definition += f", {p}: str" + + if "perPage" in parse_params(operation, parameters): if operation in REVERSE_PAGINATION: definition += ", total_pages=1, direction='prev'" else: definition += ", total_pages=1, direction='next'" - if operation == 'getNetworkEvents': - definition += ', event_log_end_time=None' + if operation == "getNetworkEvents": + definition += ", event_log_end_time=None" - if parse_params(operation, parameters, ['optional']): - definition += f', **kwargs' + if parse_params(operation, parameters, ["optional"]): + definition += f", **kwargs" # Docstring param_descriptions = list() - all_params = parse_params(operation, parameters, ['required', 'pagination', 'optional']) + all_params = parse_params( + operation, parameters, ["required", "pagination", "optional"] + ) if all_params: for p, values in all_params.items(): - param_descriptions.append(f'{p} ({values["type"]}): {values["description"]}') + param_descriptions.append( + f'{p} ({values["type"]}): {values["description"]}' + ) # Combine keyword args with locals - kwarg_line = '' - if parse_params(operation, parameters, ['optional']): - kwarg_line = 'kwargs.update(locals())' + kwarg_line = "" + if parse_params(operation, parameters, ["optional"]): + kwarg_line = "kwargs.update(locals())" # will need update for OASv3 - elif parse_params(operation, parameters, ['query', 'array', 'body']): - kwarg_line = 'kwargs = locals()' + elif parse_params(operation, parameters, ["query", "array", "body"]): + kwarg_line = "kwargs = locals()" # Assert valid values for enum - enum_params = parse_params(operation, parameters, ['enum']) + enum_params = parse_params(operation, parameters, ["enum"]) assert_blocks = list() if enum_params: for p, values in enum_params.items(): - assert_blocks.append((p, values['enum'])) + assert_blocks.append((p, values["enum"])) # Function body for GET endpoints query_params = array_params = body_params = {} - # Function body for POST/PUT endpoints - if method == 'post' or method == 'put': - - # will need update for OASv3 - body_params = parse_params(operation, parameters, 'body') - if method == 'post': - batch_operation = 'create' - else: - batch_operation = 'update' - - # Function body for DELETE endpoints - elif method == 'delete': - batch_operation = 'destroy' - # Function return statement - call_line = 'return action' + call_line = "return action" # Add function to files - with open(f'{template_dir}batch_function_template.jinja2', encoding='utf-8', newline=None) \ - as fp: + with open( + f"{template_dir}batch_function_template.jinja2", + encoding="utf-8", + newline=None, + ) as fp: function_template = fp.read() template = jinja_env.from_string(function_template) batch_output.write( - '\n\n' + - template.render( + "\n\n" + + template.render( operation=operation, function_definition=definition, description=description, @@ -582,14 +700,19 @@ def generate_action_batch_functions(jinja_env: jinja2.Environment, template_dir: array_params=array_params, body_params=body_params, call_line=call_line, - batch_operation=batch_operation + batch_operation=batch_operation, ) ) -def render_class_template(jinja_env: jinja2.Environment, template_dir: str, template_name: str, - output: open, scope: str): - with open(f'{template_dir}{template_name}', encoding='utf-8', newline=None) as fp: +def render_class_template( + jinja_env: jinja2.Environment, + template_dir: str, + template_name: str, + output: open, + scope: str, +): + with open(f"{template_dir}{template_name}", encoding="utf-8", newline=None) as fp: class_template = fp.read() template = jinja_env.from_string(class_template) output.write( @@ -601,35 +724,35 @@ def render_class_template(jinja_env: jinja2.Environment, template_dir: str, temp # Prints READ_ME help message for user to read def print_help(): - lines = READ_ME.split('\n') + lines = READ_ME.split("\n") for line in lines: - print(f'# {line}') + print(f"# {line}") # Parse command line arguments def main(inputs): - api_key = os.environ.get('MERAKI_DASHBOARD_API_KEY') + api_key = os.environ.get("MERAKI_DASHBOARD_API_KEY") org_id = None - version_number = 'custom' + version_number = "custom" is_github_action = False try: - opts, args = getopt.getopt(inputs, 'ho:k:v:g:') + opts, args = getopt.getopt(inputs, "ho:k:v:g:") except getopt.GetoptError: print_help() sys.exit(2) for opt, arg in opts: - if opt == '-h': + if opt == "-h": print_help() sys.exit(2) - elif opt == '-o': + elif opt == "-o": org_id = arg - elif opt == '-k' and api_key is None: + elif opt == "-k" and api_key is None: api_key = arg - elif opt == '-v': + elif opt == "-v": version_number = arg - elif opt == '-g': - if arg.lower() == 'true': + elif opt == "-g": + if arg.lower() == "true": is_github_action = True check_python_version() @@ -640,27 +763,31 @@ def main(inputs): print_help() sys.exit(2) else: - response = requests.get(f'https://api.meraki.com/api/v1/organizations/{org_id}/openapiSpec', - headers={'Authorization': f'Bearer {api_key}'}) + response = requests.get( + f"https://api.meraki.com/api/v1/organizations/{org_id}/openapiSpec", + headers={"Authorization": f"Bearer {api_key}"}, + ) if response.ok: spec = response.json() else: print_help() - sys.exit(f'API key provided does not have access to org {org_id}') + sys.exit(f"API key provided does not have access to org {org_id}") else: - response = requests.get('https://api.meraki.com/api/v1/openapiSpec') + response = requests.get("https://api.meraki.com/api/v1/openapiSpec") # Validate that the spec pulled successfully before trying to generate the library. if response.ok: spec = response.json() - print(f'Successfully pulled Meraki dashboard API OpenAPI spec.') + print(f"Successfully pulled Meraki dashboard API OpenAPI spec.") else: print_help() - sys.exit(f'There was an HTTP error pulling the OpenAPI specification. Please try again in a few minutes. ' - f'If this continues for more than an hour, please contact Meraki support and mention that ' - f'"HTTP GET https://api.meraki.com/api/v1/openapiSpec" is failing.') + sys.exit( + f"There was an HTTP error pulling the OpenAPI specification. Please try again in a few minutes. " + f"If this continues for more than an hour, please contact Meraki support and mention that " + f'"HTTP GET https://api.meraki.com/api/v1/openapiSpec" is failing.' + ) generate_library(spec, version_number, is_github_action) -if __name__ == '__main__': +if __name__ == "__main__": main(sys.argv[1:]) From b669f8b2c17a017e3c5ea785e57b241d0f48b3c6 Mon Sep 17 00:00:00 2001 From: "John M. Kuchta" Date: Tue, 16 Jan 2024 10:59:43 -0800 Subject: [PATCH 2/3] fix PR --- generator/generate_library.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/generator/generate_library.py b/generator/generate_library.py index ffda0f9b..f7791ec8 100644 --- a/generator/generate_library.py +++ b/generator/generate_library.py @@ -598,6 +598,10 @@ def generate_action_batch_functions( ][0] batch_operation = this_action["operation"] + + # Function body for GET endpoints + query_params = array_params = body_params = {} + # Function body for POST/PUT endpoints if method == "post" or method == "put": # will need update for OASv3 @@ -669,9 +673,6 @@ def generate_action_batch_functions( for p, values in enum_params.items(): assert_blocks.append((p, values["enum"])) - # Function body for GET endpoints - query_params = array_params = body_params = {} - # Function return statement call_line = "return action" From 42d0584d243c977cb4592891fb84961f85414d71 Mon Sep 17 00:00:00 2001 From: "John M. Kuchta" Date: Tue, 16 Jan 2024 11:07:51 -0800 Subject: [PATCH 3/3] Fix PR --- generator/generate_library.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/generator/generate_library.py b/generator/generate_library.py index f7791ec8..0b3a3979 100644 --- a/generator/generate_library.py +++ b/generator/generate_library.py @@ -599,6 +599,11 @@ def generate_action_batch_functions( batch_operation = this_action["operation"] + # May need update for OASv3 + parameters = ( + endpoint["parameters"] if "parameters" in endpoint else None + ) + # Function body for GET endpoints query_params = array_params = body_params = {} @@ -609,11 +614,6 @@ def generate_action_batch_functions( # Function body for DELETE endpoints is empty (HTTP 204) - # May need update for OASv3 - parameters = ( - endpoint["parameters"] if "parameters" in endpoint else None - ) - # Function definition definition = "" if parameters: