Initial release of new action batch functions

The new action batch functions return batch-formatted actions instead of executing the associated changes. Then you can chain the actions together and create an action batch more easily.

Author: TKIPisalegacycipher

generator/batch_class_template.jinja2

@@ -0,0 +1,3 @@
+class ActionBatch{{ class_name }}(object):
+	def __init__(self):
+		super(ActionBatch{{ class_name }}, self).__init__()

generator/batch_function_template.jinja2

@@ -0,0 +1,56 @@
+	def {{ operation }}(self{% if function_definition|length > 0 %}{{ function_definition }}{% endif %}):
+		"""
+		**{{ description }}**
+		{{ doc_url }}
+		
+		{% for d in descriptions %}
+		- {{ d }}
+		{% endfor %}
+		"""
+
+		{% if kwarg_line|length > 0 %}
+		{{ kwarg_line }}
+
+		{% endif %}
+		{% if assert_blocks|length > 0 %}
+		{% for param, values in assert_blocks %}
+		if '{{ param }}' in kwargs:
+			options = {{ values }}
+			assert kwargs['{{ param }}'] in options, f'''"{{ param }}" cannot be "{kwargs['{{ param }}']}", & must be set to one of: {options}'''
+		{% endfor %}
+
+		{% endif %}
+		metadata = {
+			'tags': {{ tags }},
+			'operation': '{{ operation }}'
+		}
+		resource = f'{{ resource }}'
+
+		{% if query_params|length > 0 %}
+		query_params = [{% for param in query_params %}'{{ param }}', {% endfor %}]
+		params = {k.strip(): v for k, v in kwargs.items() if k.strip() in query_params}
+
+		{% endif %}
+		{% if array_params|length > 0 %}
+		array_params = [{% for param in array_params %}'{{ param }}', {% endfor %}]
+		for k, v in kwargs.items():
+			if k.strip() in array_params:
+				params[f'{k.strip()}[]'] = kwargs[f'{k}']
+				params.pop(k.strip())
+
+		{% endif %}
+		{% if body_params|length > 0 %}
+		body_params = [{% for param in body_params %}'{{ param }}', {% endfor %}]
+		payload = {k.strip(): v for k, v in kwargs.items() if k.strip() in body_params}
+		{% endif %}
+		action = {
+			"resource": resource,
+			"operation": "{{ batch_operation }}",
+			"body": payload
+		}
+		{{ call_line }}
+
+
+
+
+

generator/generate_library.py

@@ -130,9 +130,10 @@ def generate_library(spec, version_number):
 	tags = spec['tags']
 	paths = spec['paths']
 	scopes = {tag['name']: {} for tag in tags[:10]}
+	batchable_action_summaries = [action['summary'] for action in spec['x-batchable-actions']]
 
 	# Check paths and create sub-directories if needed
-	subdirs = ['meraki', 'meraki/api', 'meraki/aio', 'meraki/aio/api']
+	subdirs = ['meraki', 'meraki/api', 'meraki/api/batch', 'meraki/aio', 'meraki/aio/api', 'meraki/api/batch']
 	for dir in subdirs:
 		if not os.path.isdir(dir):
 			os.mkdir(dir)
@@ -182,6 +183,7 @@ def generate_library(spec, version_number):
 					)
 				)
 
+			# Generate Asyncio API libraries
 			async_output = open(f'meraki/aio/api/{scope}.py', 'w', encoding='utf-8')
 			with open('async_class_template.jinja2', encoding='utf-8') as fp:
 				class_template = fp.read()
@@ -192,6 +194,18 @@ def generate_library(spec, version_number):
 					)
 				)
 
+			# Generate Action Batch API libraries
+			batch_output = open(f'meraki/api/batch/{scope}.py', 'w', encoding='utf-8')
+			with open('batch_class_template.jinja2', encoding='utf-8') as fp:
+				class_template = fp.read()
+				template = jinja_env.from_string(class_template)
+				batch_output.write(
+					template.render(
+						class_name=scope[0].upper() + scope[1:],
+					)
+				)
+
+			# Generate API & Asyncio API functions
 			for path, methods in section.items():
 				for method, endpoint in methods.items():
 					# Get metadata
@@ -299,7 +313,7 @@ def generate_library(spec, version_number):
 								query_params=query_params,
 								array_params=array_params,
 								body_params=body_params,
-								call_line=call_line,
+								call_line=call_line
 							)
 						)
 						async_output.write(
@@ -318,10 +332,113 @@ def generate_library(spec, version_number):
 								query_params=query_params,
 								array_params=array_params,
 								body_params=body_params,
-								call_line=call_line,
+								call_line=call_line
 							)
 						)
 
+			# Generate API action batch functions
+			for path, methods in section.items():
+				for method, endpoint in methods.items():
+					if endpoint['description'] in batchable_action_summaries:
+						# Get metadata
+						tags = endpoint['tags']
+						operation = endpoint['operationId']
+						description = endpoint['summary']
+						parameters = endpoint['parameters'] if 'parameters' in endpoint else None
+						responses = endpoint['responses']  # not actually used here for library generation
+
+						# Function 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):
+								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 parse_params(operation, parameters, ['optional']):
+								definition += f', **kwargs'
+
+						# Docstring
+						param_descriptions = []
+						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"]}')
+
+						# 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()'
+
+						# Assert valid values for enum
+						enum_params = parse_params(operation, parameters, ['enum'])
+						assert_blocks = []
+						if enum_params:
+							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 body for POST/PUT endpoints
+						if method == 'post' or method == 'put':
+							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'
+
+						# Add function to files
+						with open('batch_function_template.jinja2', encoding='utf-8') as fp:
+							function_template = fp.read()
+							template = jinja_env.from_string(function_template)
+							batch_output.write(
+								'\n\n' +
+								template.render(
+									operation=operation,
+									function_definition=definition,
+									description=description,
+									doc_url=docs_url(operation),
+									descriptions=param_descriptions,
+									kwarg_line=kwarg_line,
+									all_params=list(all_params.keys()),
+									assert_blocks=assert_blocks,
+									tags=tags,
+									resource=path,
+									query_params=query_params,
+									array_params=array_params,
+									body_params=body_params,
+									call_line=call_line,
+									batch_operation=batch_operation
+								)
+							)
+
 
 # Prints READ_ME help message for user to read
 def print_help():

meraki/__init__.py

@@ -13,14 +13,29 @@
 from .api.sm import Sm
 from .api.switch import Switch
 from .api.wireless import Wireless
+
+# Batch class imports
+from .api.batch.organizations import ActionBatchOrganizations
+from .api.batch.networks import ActionBatchNetworks
+from .api.batch.devices import ActionBatchDevices
+from .api.batch.appliance import ActionBatchAppliance
+from .api.batch.camera import ActionBatchCamera
+from .api.batch.cellularGateway import ActionBatchCellularGateway
+from .api.batch.insight import ActionBatchInsight
+from .api.batch.sm import ActionBatchSm
+from .api.batch.switch import ActionBatchSwitch
+from .api.batch.wireless import ActionBatchWireless
+
+# Config import
 from .config import (
     API_KEY_ENVIRONMENT_VARIABLE, DEFAULT_BASE_URL, SINGLE_REQUEST_TIMEOUT, CERTIFICATE_PATH, REQUESTS_PROXY,
     WAIT_ON_RATE_LIMIT, NGINX_429_RETRY_WAIT_TIME, ACTION_BATCH_RETRY_WAIT_TIME, RETRY_4XX_ERROR,
     RETRY_4XX_ERROR_WAIT_TIME, MAXIMUM_RETRIES, OUTPUT_LOG, LOG_PATH, LOG_FILE_PREFIX, PRINT_TO_CONSOLE,
     SUPPRESS_LOGGING, SIMULATE_API_CALLS, BE_GEO_ID, MERAKI_PYTHON_SDK_CALLER
 )
 
-__version__ = '1.6.2'
+__version__ = '1.6.3a'
+
 
 class DashboardAPI(object):
     """
@@ -129,3 +144,23 @@ def __init__(self, api_key=None, base_url=DEFAULT_BASE_URL, single_request_timeo
         self.sm = Sm(self._session)
         self.switch = Switch(self._session)
         self.wireless = Wireless(self._session)
+
+        # Batch class
+        class Batch:
+            def __init__(self):
+                pass
+
+        # Batch definitions
+        self.batch = Batch()
+
+        # Action Batch API endpoints by section
+        self.batch.organizations = ActionBatchOrganizations()
+        self.batch.networks = ActionBatchNetworks()
+        self.batch.devices = ActionBatchDevices()
+        self.batch.appliance = ActionBatchAppliance()
+        self.batch.camera = ActionBatchCamera()
+        self.batch.cellularGateway = ActionBatchCellularGateway()
+        self.batch.insight = ActionBatchInsight()
+        self.batch.sm = ActionBatchSm()
+        self.batch.switch = ActionBatchSwitch()
+        self.batch.wireless = ActionBatchWireless()