Best Python code snippet using slash
lambda_facts.py
Source:lambda_facts.py
1#!/usr/bin/python2# This file is part of Ansible3# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)4from __future__ import absolute_import, division, print_function5__metaclass__ = type6DOCUMENTATION = '''7---8module: lambda_facts9version_added: 1.0.010deprecated:11 removed_at_date: '2021-12-01'12 removed_from_collection: 'community.aws'13 why: Deprecated in favour of C(_info) module.14 alternative: Use M(community.aws.lambda_info) instead.15short_description: Gathers AWS Lambda function details as Ansible facts16description:17 - Gathers various details related to Lambda functions, including aliases, versions and event source mappings.18 Use module M(community.aws.lambda) to manage the lambda function itself, M(community.aws.lambda_alias) to manage function aliases and19 M(community.aws.lambda_event) to manage lambda event source mappings.20options:21 query:22 description:23 - Specifies the resource type for which to gather facts. Leave blank to retrieve all facts.24 choices: [ "aliases", "all", "config", "mappings", "policy", "versions" ]25 default: "all"26 type: str27 function_name:28 description:29 - The name of the lambda function for which facts are requested.30 aliases: [ "function", "name"]31 type: str32 event_source_arn:33 description:34 - For query type 'mappings', this is the Amazon Resource Name (ARN) of the Amazon Kinesis or DynamoDB stream.35 type: str36author: Pierre Jodouin (@pjodouin)37requirements:38 - boto339extends_documentation_fragment:40- amazon.aws.aws41- amazon.aws.ec242'''43EXAMPLES = '''44---45# Simple example of listing all info for a function46- name: List all for a specific function47 community.aws.lambda_facts:48 query: all49 function_name: myFunction50 register: my_function_details51# List all versions of a function52- name: List function versions53 community.aws.lambda_facts:54 query: versions55 function_name: myFunction56 register: my_function_versions57# List all lambda function versions58- name: List all function59 community.aws.lambda_facts:60 query: all61 max_items: 2062- name: show Lambda facts63 ansible.builtin.debug:64 var: lambda_facts65'''66RETURN = '''67---68lambda_facts:69 description: lambda facts70 returned: success71 type: dict72lambda_facts.function:73 description: lambda function list74 returned: success75 type: dict76lambda_facts.function.TheName:77 description: lambda function information, including event, mapping, and version information78 returned: success79 type: dict80'''81from ansible_collections.amazon.aws.plugins.module_utils.core import AnsibleAWSModule82from ansible_collections.amazon.aws.plugins.module_utils.ec2 import camel_dict_to_snake_dict83import json84import datetime85import sys86import re87try:88 from botocore.exceptions import ClientError89except ImportError:90 pass # caught by AnsibleAWSModule91def fix_return(node):92 """93 fixup returned dictionary94 :param node:95 :return:96 """97 if isinstance(node, datetime.datetime):98 node_value = str(node)99 elif isinstance(node, list):100 node_value = [fix_return(item) for item in node]101 elif isinstance(node, dict):102 node_value = dict([(item, fix_return(node[item])) for item in node.keys()])103 else:104 node_value = node105 return node_value106def alias_details(client, module):107 """108 Returns list of aliases for a specified function.109 :param client: AWS API client reference (boto3)110 :param module: Ansible module reference111 :return dict:112 """113 lambda_facts = dict()114 function_name = module.params.get('function_name')115 if function_name:116 params = dict()117 if module.params.get('max_items'):118 params['MaxItems'] = module.params.get('max_items')119 if module.params.get('next_marker'):120 params['Marker'] = module.params.get('next_marker')121 try:122 lambda_facts.update(aliases=client.list_aliases(FunctionName=function_name, **params)['Aliases'])123 except ClientError as e:124 if e.response['Error']['Code'] == 'ResourceNotFoundException':125 lambda_facts.update(aliases=[])126 else:127 module.fail_json_aws(e, msg="Trying to get aliases")128 else:129 module.fail_json(msg='Parameter function_name required for query=aliases.')130 return {function_name: camel_dict_to_snake_dict(lambda_facts)}131def all_details(client, module):132 """133 Returns all lambda related facts.134 :param client: AWS API client reference (boto3)135 :param module: Ansible module reference136 :return dict:137 """138 if module.params.get('max_items') or module.params.get('next_marker'):139 module.fail_json(msg='Cannot specify max_items nor next_marker for query=all.')140 lambda_facts = dict()141 function_name = module.params.get('function_name')142 if function_name:143 lambda_facts[function_name] = {}144 lambda_facts[function_name].update(config_details(client, module)[function_name])145 lambda_facts[function_name].update(alias_details(client, module)[function_name])146 lambda_facts[function_name].update(policy_details(client, module)[function_name])147 lambda_facts[function_name].update(version_details(client, module)[function_name])148 lambda_facts[function_name].update(mapping_details(client, module)[function_name])149 else:150 lambda_facts.update(config_details(client, module))151 return lambda_facts152def config_details(client, module):153 """154 Returns configuration details for one or all lambda functions.155 :param client: AWS API client reference (boto3)156 :param module: Ansible module reference157 :return dict:158 """159 lambda_facts = dict()160 function_name = module.params.get('function_name')161 if function_name:162 try:163 lambda_facts.update(client.get_function_configuration(FunctionName=function_name))164 except ClientError as e:165 if e.response['Error']['Code'] == 'ResourceNotFoundException':166 lambda_facts.update(function={})167 else:168 module.fail_json_aws(e, msg="Trying to get {0} configuration".format(function_name))169 else:170 params = dict()171 if module.params.get('max_items'):172 params['MaxItems'] = module.params.get('max_items')173 if module.params.get('next_marker'):174 params['Marker'] = module.params.get('next_marker')175 try:176 lambda_facts.update(function_list=client.list_functions(**params)['Functions'])177 except ClientError as e:178 if e.response['Error']['Code'] == 'ResourceNotFoundException':179 lambda_facts.update(function_list=[])180 else:181 module.fail_json_aws(e, msg="Trying to get function list")182 functions = dict()183 for func in lambda_facts.pop('function_list', []):184 functions[func['FunctionName']] = camel_dict_to_snake_dict(func)185 return functions186 return {function_name: camel_dict_to_snake_dict(lambda_facts)}187def mapping_details(client, module):188 """189 Returns all lambda event source mappings.190 :param client: AWS API client reference (boto3)191 :param module: Ansible module reference192 :return dict:193 """194 lambda_facts = dict()195 params = dict()196 function_name = module.params.get('function_name')197 if function_name:198 params['FunctionName'] = module.params.get('function_name')199 if module.params.get('event_source_arn'):200 params['EventSourceArn'] = module.params.get('event_source_arn')201 if module.params.get('max_items'):202 params['MaxItems'] = module.params.get('max_items')203 if module.params.get('next_marker'):204 params['Marker'] = module.params.get('next_marker')205 try:206 lambda_facts.update(mappings=client.list_event_source_mappings(**params)['EventSourceMappings'])207 except ClientError as e:208 if e.response['Error']['Code'] == 'ResourceNotFoundException':209 lambda_facts.update(mappings=[])210 else:211 module.fail_json_aws(e, msg="Trying to get source event mappings")212 if function_name:213 return {function_name: camel_dict_to_snake_dict(lambda_facts)}214 return camel_dict_to_snake_dict(lambda_facts)215def policy_details(client, module):216 """217 Returns policy attached to a lambda function.218 :param client: AWS API client reference (boto3)219 :param module: Ansible module reference220 :return dict:221 """222 if module.params.get('max_items') or module.params.get('next_marker'):223 module.fail_json(msg='Cannot specify max_items nor next_marker for query=policy.')224 lambda_facts = dict()225 function_name = module.params.get('function_name')226 if function_name:227 try:228 # get_policy returns a JSON string so must convert to dict before reassigning to its key229 lambda_facts.update(policy=json.loads(client.get_policy(FunctionName=function_name)['Policy']))230 except ClientError as e:231 if e.response['Error']['Code'] == 'ResourceNotFoundException':232 lambda_facts.update(policy={})233 else:234 module.fail_json_aws(e, msg="Trying to get {0} policy".format(function_name))235 else:236 module.fail_json(msg='Parameter function_name required for query=policy.')237 return {function_name: camel_dict_to_snake_dict(lambda_facts)}238def version_details(client, module):239 """240 Returns all lambda function versions.241 :param client: AWS API client reference (boto3)242 :param module: Ansible module reference243 :return dict:244 """245 lambda_facts = dict()246 function_name = module.params.get('function_name')247 if function_name:248 params = dict()249 if module.params.get('max_items'):250 params['MaxItems'] = module.params.get('max_items')251 if module.params.get('next_marker'):252 params['Marker'] = module.params.get('next_marker')253 try:254 lambda_facts.update(versions=client.list_versions_by_function(FunctionName=function_name, **params)['Versions'])255 except ClientError as e:256 if e.response['Error']['Code'] == 'ResourceNotFoundException':257 lambda_facts.update(versions=[])258 else:259 module.fail_json_aws(e, msg="Trying to get {0} versions".format(function_name))260 else:261 module.fail_json(msg='Parameter function_name required for query=versions.')262 return {function_name: camel_dict_to_snake_dict(lambda_facts)}263def main():264 """265 Main entry point.266 :return dict: ansible facts267 """268 argument_spec = dict(269 function_name=dict(required=False, default=None, aliases=['function', 'name']),270 query=dict(required=False, choices=['aliases', 'all', 'config', 'mappings', 'policy', 'versions'], default='all'),271 event_source_arn=dict(required=False, default=None)272 )273 module = AnsibleAWSModule(274 argument_spec=argument_spec,275 supports_check_mode=True,276 mutually_exclusive=[],277 required_together=[]278 )279 # validate function_name if present280 function_name = module.params['function_name']281 if function_name:282 if not re.search(r"^[\w\-:]+$", function_name):283 module.fail_json(284 msg='Function name {0} is invalid. Names must contain only alphanumeric characters and hyphens.'.format(function_name)285 )286 if len(function_name) > 64:287 module.fail_json(msg='Function name "{0}" exceeds 64 character limit'.format(function_name))288 client = module.client('lambda')289 this_module = sys.modules[__name__]290 invocations = dict(291 aliases='alias_details',292 all='all_details',293 config='config_details',294 mappings='mapping_details',295 policy='policy_details',296 versions='version_details',297 )298 this_module_function = getattr(this_module, invocations[module.params['query']])299 all_facts = fix_return(this_module_function(client, module))300 results = dict(ansible_facts={'lambda_facts': {'function': all_facts}}, changed=False)301 if module.check_mode:302 results['msg'] = 'Check mode set but ignored for fact gathering only.'303 module.exit_json(**results)304if __name__ == '__main__':...
lambda_info.py
Source:lambda_info.py
1#!/usr/bin/python2# This file is part of Ansible3# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)4from __future__ import absolute_import, division, print_function5__metaclass__ = type6DOCUMENTATION = '''7---8module: lambda_info9version_added: 1.0.010short_description: Gathers AWS Lambda function details11description:12 - Gathers various details related to Lambda functions, including aliases, versions and event source mappings.13 - Use module M(community.aws.lambda) to manage the lambda function itself, M(community.aws.lambda_alias) to manage function aliases and14 M(community.aws.lambda_event) to manage lambda event source mappings.15options:16 query:17 description:18 - Specifies the resource type for which to gather information. Leave blank to retrieve all information.19 choices: [ "aliases", "all", "config", "mappings", "policy", "versions" ]20 default: "all"21 type: str22 function_name:23 description:24 - The name of the lambda function for which information is requested.25 aliases: [ "function", "name"]26 type: str27 event_source_arn:28 description:29 - When I(query=mappings), this is the Amazon Resource Name (ARN) of the Amazon Kinesis or DynamoDB stream.30 type: str31author: Pierre Jodouin (@pjodouin)32requirements:33 - boto334extends_documentation_fragment:35- amazon.aws.aws36- amazon.aws.ec237'''38EXAMPLES = '''39---40# Simple example of listing all info for a function41- name: List all for a specific function42 community.aws.lambda_info:43 query: all44 function_name: myFunction45 register: my_function_details46# List all versions of a function47- name: List function versions48 community.aws.lambda_info:49 query: versions50 function_name: myFunction51 register: my_function_versions52# List all lambda function versions53- name: List all function54 community.aws.lambda_info:55 query: all56 max_items: 2057 register: output58- name: show Lambda information59 ansible.builtin.debug:60 msg: "{{ output['function'] }}"61'''62RETURN = '''63---64function:65 description: lambda function list66 returned: success67 type: dict68function.TheName:69 description: lambda function information, including event, mapping, and version information70 returned: success71 type: dict72'''73from ansible_collections.amazon.aws.plugins.module_utils.core import AnsibleAWSModule74from ansible_collections.amazon.aws.plugins.module_utils.ec2 import camel_dict_to_snake_dict75import json76import datetime77import re78try:79 from botocore.exceptions import ClientError80except ImportError:81 pass # caught by AnsibleAWSModule82def fix_return(node):83 """84 fixup returned dictionary85 :param node:86 :return:87 """88 if isinstance(node, datetime.datetime):89 node_value = str(node)90 elif isinstance(node, list):91 node_value = [fix_return(item) for item in node]92 elif isinstance(node, dict):93 node_value = dict([(item, fix_return(node[item])) for item in node.keys()])94 else:95 node_value = node96 return node_value97def alias_details(client, module):98 """99 Returns list of aliases for a specified function.100 :param client: AWS API client reference (boto3)101 :param module: Ansible module reference102 :return dict:103 """104 lambda_info = dict()105 function_name = module.params.get('function_name')106 if function_name:107 params = dict()108 if module.params.get('max_items'):109 params['MaxItems'] = module.params.get('max_items')110 if module.params.get('next_marker'):111 params['Marker'] = module.params.get('next_marker')112 try:113 lambda_info.update(aliases=client.list_aliases(FunctionName=function_name, **params)['Aliases'])114 except ClientError as e:115 if e.response['Error']['Code'] == 'ResourceNotFoundException':116 lambda_info.update(aliases=[])117 else:118 module.fail_json_aws(e, msg="Trying to get aliases")119 else:120 module.fail_json(msg='Parameter function_name required for query=aliases.')121 return {function_name: camel_dict_to_snake_dict(lambda_info)}122def all_details(client, module):123 """124 Returns all lambda related facts.125 :param client: AWS API client reference (boto3)126 :param module: Ansible module reference127 :return dict:128 """129 if module.params.get('max_items') or module.params.get('next_marker'):130 module.fail_json(msg='Cannot specify max_items nor next_marker for query=all.')131 lambda_info = dict()132 function_name = module.params.get('function_name')133 if function_name:134 lambda_info[function_name] = {}135 lambda_info[function_name].update(config_details(client, module)[function_name])136 lambda_info[function_name].update(alias_details(client, module)[function_name])137 lambda_info[function_name].update(policy_details(client, module)[function_name])138 lambda_info[function_name].update(version_details(client, module)[function_name])139 lambda_info[function_name].update(mapping_details(client, module)[function_name])140 else:141 lambda_info.update(config_details(client, module))142 return lambda_info143def config_details(client, module):144 """145 Returns configuration details for one or all lambda functions.146 :param client: AWS API client reference (boto3)147 :param module: Ansible module reference148 :return dict:149 """150 lambda_info = dict()151 function_name = module.params.get('function_name')152 if function_name:153 try:154 lambda_info.update(client.get_function_configuration(FunctionName=function_name))155 except ClientError as e:156 if e.response['Error']['Code'] == 'ResourceNotFoundException':157 lambda_info.update(function={})158 else:159 module.fail_json_aws(e, msg="Trying to get {0} configuration".format(function_name))160 else:161 params = dict()162 if module.params.get('max_items'):163 params['MaxItems'] = module.params.get('max_items')164 if module.params.get('next_marker'):165 params['Marker'] = module.params.get('next_marker')166 try:167 lambda_info.update(function_list=client.list_functions(**params)['Functions'])168 except ClientError as e:169 if e.response['Error']['Code'] == 'ResourceNotFoundException':170 lambda_info.update(function_list=[])171 else:172 module.fail_json_aws(e, msg="Trying to get function list")173 functions = dict()174 for func in lambda_info.pop('function_list', []):175 functions[func['FunctionName']] = camel_dict_to_snake_dict(func)176 return functions177 return {function_name: camel_dict_to_snake_dict(lambda_info)}178def mapping_details(client, module):179 """180 Returns all lambda event source mappings.181 :param client: AWS API client reference (boto3)182 :param module: Ansible module reference183 :return dict:184 """185 lambda_info = dict()186 params = dict()187 function_name = module.params.get('function_name')188 if function_name:189 params['FunctionName'] = module.params.get('function_name')190 if module.params.get('event_source_arn'):191 params['EventSourceArn'] = module.params.get('event_source_arn')192 if module.params.get('max_items'):193 params['MaxItems'] = module.params.get('max_items')194 if module.params.get('next_marker'):195 params['Marker'] = module.params.get('next_marker')196 try:197 lambda_info.update(mappings=client.list_event_source_mappings(**params)['EventSourceMappings'])198 except ClientError as e:199 if e.response['Error']['Code'] == 'ResourceNotFoundException':200 lambda_info.update(mappings=[])201 else:202 module.fail_json_aws(e, msg="Trying to get source event mappings")203 if function_name:204 return {function_name: camel_dict_to_snake_dict(lambda_info)}205 return camel_dict_to_snake_dict(lambda_info)206def policy_details(client, module):207 """208 Returns policy attached to a lambda function.209 :param client: AWS API client reference (boto3)210 :param module: Ansible module reference211 :return dict:212 """213 if module.params.get('max_items') or module.params.get('next_marker'):214 module.fail_json(msg='Cannot specify max_items nor next_marker for query=policy.')215 lambda_info = dict()216 function_name = module.params.get('function_name')217 if function_name:218 try:219 # get_policy returns a JSON string so must convert to dict before reassigning to its key220 lambda_info.update(policy=json.loads(client.get_policy(FunctionName=function_name)['Policy']))221 except ClientError as e:222 if e.response['Error']['Code'] == 'ResourceNotFoundException':223 lambda_info.update(policy={})224 else:225 module.fail_json_aws(e, msg="Trying to get {0} policy".format(function_name))226 else:227 module.fail_json(msg='Parameter function_name required for query=policy.')228 return {function_name: camel_dict_to_snake_dict(lambda_info)}229def version_details(client, module):230 """231 Returns all lambda function versions.232 :param client: AWS API client reference (boto3)233 :param module: Ansible module reference234 :return dict:235 """236 lambda_info = dict()237 function_name = module.params.get('function_name')238 if function_name:239 params = dict()240 if module.params.get('max_items'):241 params['MaxItems'] = module.params.get('max_items')242 if module.params.get('next_marker'):243 params['Marker'] = module.params.get('next_marker')244 try:245 lambda_info.update(versions=client.list_versions_by_function(FunctionName=function_name, **params)['Versions'])246 except ClientError as e:247 if e.response['Error']['Code'] == 'ResourceNotFoundException':248 lambda_info.update(versions=[])249 else:250 module.fail_json_aws(e, msg="Trying to get {0} versions".format(function_name))251 else:252 module.fail_json(msg='Parameter function_name required for query=versions.')253 return {function_name: camel_dict_to_snake_dict(lambda_info)}254def main():255 """256 Main entry point.257 :return dict: ansible facts258 """259 argument_spec = dict(260 function_name=dict(required=False, default=None, aliases=['function', 'name']),261 query=dict(required=False, choices=['aliases', 'all', 'config', 'mappings', 'policy', 'versions'], default='all'),262 event_source_arn=dict(required=False, default=None)263 )264 module = AnsibleAWSModule(265 argument_spec=argument_spec,266 supports_check_mode=True,267 mutually_exclusive=[],268 required_together=[]269 )270 # validate function_name if present271 function_name = module.params['function_name']272 if function_name:273 if not re.search(r"^[\w\-:]+$", function_name):274 module.fail_json(275 msg='Function name {0} is invalid. Names must contain only alphanumeric characters and hyphens.'.format(function_name)276 )277 if len(function_name) > 64:278 module.fail_json(msg='Function name "{0}" exceeds 64 character limit'.format(function_name))279 client = module.client('lambda')280 invocations = dict(281 aliases='alias_details',282 all='all_details',283 config='config_details',284 mappings='mapping_details',285 policy='policy_details',286 versions='version_details',287 )288 this_module_function = globals()[invocations[module.params['query']]]289 all_facts = fix_return(this_module_function(client, module))290 results = dict(function=all_facts, changed=False)291 if module.check_mode:292 results['msg'] = 'Check mode set but ignored for fact gathering only.'293 module.exit_json(**results)294if __name__ == '__main__':...
docstring_errors.py
Source:docstring_errors.py
1from pathlib import Path2class DocstringError:3 def __init__(self, file_path: Path, function_name: str):4 self.function_name = function_name5 self.file_path = file_path6 # line_number: int # TODO: implement7class IncorrectArgumentTypehintError(DocstringError):8 def __init__(self, file_path: Path, function_name: str, argument_name: str):9 super().__init__(file_path, function_name)10 self.argument_name = argument_name11 def get_message(self) -> str:12 return f"Function `{self.function_name}` argument `{self.argument_name}` has the wrong type hint."13class IncorrectReturnTypehintError(DocstringError):14 def __init__(self, file_path: Path, function_name: str):15 super().__init__(file_path, function_name)16 def get_message(self) -> str:17 return ""18class MissingArgsSectionError(DocstringError):19 def __init__(self, file_path: Path, function_name: str):20 super().__init__(file_path, function_name)21 def get_message(self) -> str:22 return ""23class MissingArgumentTypehintError(DocstringError):24 def __init__(self, file_path: Path, function_name: str, argument_name: str):25 super().__init__(file_path, function_name)26 self.argument_name = argument_name27 def get_message(self) -> str:28 return f"Function `{self.function_name}` argument `{self.argument_name}` did not have a type hint."29class MissingArgumentError(DocstringError):30 def __init__(self, file_path: Path, function_name: str, argument_name: str):31 super().__init__(file_path, function_name)32 self.argument_name = argument_name33 def get_message(self) -> str:34 return f"Function `{self.function_name}` argument `{self.argument_name}` did not have a type hint."35class MissingDocstringError(DocstringError):36 def __init__(self, file_path: Path, function_name: str):37 super().__init__(file_path, function_name)38 def get_message(self) -> str:39 return f"Function `{self.function_name}` is missing a docstring."40class MissingReturnTypehintError(DocstringError):41 def __init__(self, file_path: Path, function_name: str):42 super().__init__(file_path, function_name)43 def get_message(self) -> str:44 return ""45class MissingReturnsSectionError(DocstringError):46 def __init__(self, file_path: Path, function_name: str):47 super().__init__(file_path, function_name)48 def get_message(self) -> str:49 return ""50class MissingSummaryError(DocstringError):51 def __init__(self, file_path: Path, function_name: str):52 super().__init__(file_path, function_name)53 def get_message(self) -> str:54 return ""55class UndocumentedArgumentError(DocstringError):56 def __init__(self, file_path: Path, function_name: str, argument_name: str):57 super().__init__(file_path, function_name)58 self.argument_name = argument_name59 def get_message(self) -> str:...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!