Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use python_function_name method in localstack

Best Python code snippet using localstack_python

generate.py

Source:generate.py

1import sys2import yaml3from jinja2 import Environment, FileSystemLoader4# from pprint import pprint5def generate(yaml_file):6    function_mapping = yaml.safe_load(open("function_mapping.yaml"))7    # pprint(function_mapping)8    openapi = yaml.safe_load(open(yaml_file))9    # pprint(openapi["paths"])10    python_functions = {}11    for path_openapi in openapi["paths"]:12        for verb in openapi["paths"][path_openapi]:13            verb_path_openapi = openapi["paths"][path_openapi][verb]14            uri = "/" + ("/".join(path_openapi.split("/")[3:]))15            # print (">"+uri)16            python_function_name = None17            if function_mapping.get(uri):18                if function_mapping[uri].get(verb):19                    python_function_name = function_mapping[uri][verb]20                    # print ("override")21            if python_function_name == "skip":22                continue23            if python_function_name is None:24                python_function_name = verb + uri.replace("{hub}", "").replace(25                    "{file}", ""26                ).replace("{loc}", "").replace("{loc_group}", "").replace(27                    "{job}", ""28                ).replace(29                    "{table}", ""30                ).replace(31                    "{task}", ""32                ).replace(33                    "{ev_id}", ""34                ).replace(35                    "{archive}", ""36                ).replace(37                    "{ctrl_id}", ""38                ).replace(39                    "{user}", ""40                ).replace(41                    "{license}", ""42                ).replace(43                    "{channel}", ""44                ).replace(45                    "{prop}", ""46                ).replace(47                    "{member}", ""48                ).replace(49                    "{alert}", ""50                ).replace(51                    "{attr}", ""52                ).replace(53                    "{var}", ""54                ).replace(55                    "/", "_"56                ).replace(57                    "__", "_"58                ).rstrip(59                    "_"60                )61            print(python_function_name)62            if python_functions.get(python_function_name):63                print("Duplicate: " + python_function_name)64            python_functions[python_function_name] = generate_function(65                python_function_name, verb, path_openapi, verb_path_openapi, openapi66            )67    env = Environment(loader=FileSystemLoader("./"))68    main_template = env.get_template("pyhvr_client.py.jinja")69    main_rendered = main_template.render(function_definitions=python_functions)70    with open("pyhvr_client.py", "w") as text_file:71        text_file.write(main_rendered)72    # print(main_rendered)73def generate_function(function_name, verb, uri, openapi, full_openapi):74    # pprint(uri)75    parameters = ["self"]76    query_builder_entries = []77    header_builder_entries = []78    if openapi.get("parameters"):79        for p in openapi["parameters"]:80            if p["in"] == "path":81                parameters.append(p["name"])82            elif p["in"] == "query":83                param = p["name"]84                schema = p["schema"]85                is_bool = schema.get("type") == "boolean"86                if is_bool:87                    value = f"self.from_bool({param})"88                else:89                    value = f"{param}"90                parameters.append(param + "=None")91                query_builder_entries.append(92                    f'        if {param}:\n          query["{param}"]={value}'93                )94            elif p["in"] == "header":95                param = p["name"]96                py_name = param.replace("-", "_").lower()97                schema = p["schema"]98                parameters.append(py_name + "=None")99                header_builder_entries.append(100                    f'        if {py_name}:\n          headers["{param}"]={py_name}'101                )102            else:103                print("Unsupported parameter type: " + p["in"])104    payload_builder = ""105    if openapi.get("requestBody"):106        schema = openapi["requestBody"]["content"]["application/json"]["schema"]107        payload_builder_entries = []108        payload_builder = "payload=''"109        if schema.get("properties"):110            schema_to_parse = schema["properties"]111            required = schema.get("required", [])112            param_passthrough = False113        elif schema.get("x-patternProperties"):114            schema_ref = schema["x-patternProperties"]["(*)"]["$ref"]115            # example: '#/components/schemas/Location'116            path = schema_ref.split("/")117            schema_reffed = full_openapi[path[1]][path[2]][path[3]]118            schema_to_parse = schema_reffed["properties"]119            required = schema_reffed.get("required", [])120            param_passthrough = False121        else:122            # print("Schema has no properties?")123            # pprint(schema)124            param_passthrough = True125        if param_passthrough:126            parameters.append("**payload")127            payload_builder = ""128        else:129            for property in schema_to_parse:130                is_bool = False131                try:132                    is_bool = schema_to_parse[property].get("type") == "boolean"133                    is_bool = schema_to_parse[property]["schema"]["type"] == "boolean"134                except KeyError:135                    pass136                if is_bool:137                    value = f"self.from_bool({property})"138                else:139                    value = f"{property}"140                if property in required:141                    parameters.append(property)142                    payload_builder_entries.append(143                        f'        payload["{property}"] = {value}'144                    )145                else:146                    parameters.append(property + "=None")147                    payload_builder_entries.append(148                        f'        if {property} is not None:\n          payload["{property}"]={value}'149                    )150            if payload_builder_entries:151                payload_builder = "        payload={}\n" + "\n".join(152                    payload_builder_entries153                )154        # pprint(schema)155        payload = "payload"156    else:157        payload = "None"158    if query_builder_entries:159        query_builder = "        query={}\n" + "\n".join(query_builder_entries)160        query = "query"161    else:162        query_builder = ""163        query = "None"164    if header_builder_entries:165        header_builder = "        headers={}\n" + "\n".join(header_builder_entries)166        header = "headers"167    else:168        header_builder = ""169        header = "{}"170    is_json = True171    if openapi.get("responses"):172        r = openapi["responses"]173        ok_resp = r.get("200", r.get("201", r.get("202", r.get("204"))))174        # pprint(ok_resp)175        if ok_resp:176            try:177                rt_payload = list(ok_resp.get("content", {}).keys())[0]178                # pprint(rt_payload)179                if "text/plain" in rt_payload:180                    is_json = False181            except IndexError:182                pass183    required_params = [184        param for param in parameters if ("=" not in param and "**" not in param)185    ]186    passthrough_params = [param for param in parameters if ("**" in param)]187    optional_params = [param for param in parameters if ("=" in param)]188    parameters_joined = ", ".join(189        required_params + optional_params + passthrough_params190    )191    # if uri is parametrized, make it so192    if "{" in uri:193        f = "f"194    else:195        f = ""196    fun = f"""197    def {function_name}({parameters_joined}):198{query_builder}199{header_builder}200{payload_builder}201        return self.{verb}(202            {f}"{uri}", {query}, {header}, {payload}, {is_json}203        )204"""205    return fun206    # print(fun)...

defaults.py

Source:defaults.py

1PRODUCT_DEFAULT_MAP = [2{'external_type': 'unicode',3 'field_model': 'product.template',4 'field_name': 'weight',5 'function_name': '',6 'mage_fieldname': 'weight',7 'mapping_type': 'direct',8 'name': 'Product',9 'type': 'in_out'},10{'external_type': 'unicode',11 'field_model': 'product.template',12 'field_name': 'standard_price',13 'function_name': '',14 'mage_fieldname': 'cost',15 'mapping_type': 'direct',16 'name': 'Product',17 'type': 'in_out'},18{'external_type': 'unicode',19 'field_model': 'product.template',20 'field_name': 'upc',21 'function_name': '',22 'mage_fieldname': 'upc',23 'mapping_type': 'direct',24 'name': 'Product',25 'type': 'in_out'},26{'external_type': 'unicode',27 'field_model': 'product.template',28 'field_name': 'list_price',29 'function_name': '',30 'mage_fieldname': 'price',31 'mapping_type': 'direct',32 'name': 'Product',33 'type': 'in_out'},34]35DEFAULT_JOBS = [36                {'name': 'Sync All Products',37                 'mapping_model_name': 'product.product',38                 'python_function_name': 'import_all_products',39                 'job_type': 'system',40		 'scheduler': False,41		 'mapping_name': 'Product',42		 'mapping_lines': PRODUCT_DEFAULT_MAP},43                {'name': 'Sync All Partners',44                 'mapping_model_name': '',45                 'python_function_name': 'import_all_partners',46                 'job_type': 'system',47                 'scheduler': False,48                 'mapping_name': '',49                 'mapping_lines': ''},50                {'name': 'Sync Updated Partners',51                 'mapping_model_name': '',52                 'python_function_name': 'import_updated_partners',53                 'job_type': 'system',54                 'scheduler': True,55                 'mapping_name': '',56                 'mapping_lines': ''},57                {'name': 'Sync Updated Products',58                 'mapping_model_name': 'product.product',59                 'python_function_name': 'import_updated_products',60                 'job_type': 'system',61		 'scheduler': True,62		 'mapping_name': 'Product',63		 'mapping_lines': PRODUCT_DEFAULT_MAP},64                {'name': 'Sync Admin Users',65                 'mapping_model_name': '',66                 'python_function_name': 'sync_admin_users',67                 'job_type': 'system',68                 'scheduler': False,69                 'mapping_name': '',70                 'mapping_lines': ''},71                {'name': 'Sync Sales Orders',72                 'mapping_model_name': 'sale.order',73                 'python_function_name': 'import_sales_orders',74                 'job_type': 'system',75		 'scheduler': True,76                 'mapping_name': 'Sale Order',77                 'mapping_lines': ''},78                {'name': 'Sync Categories',79                 'mapping_model_name': 'product.category',80                 'python_function_name': 'import_categories',81                 'job_type': 'system',82                 'scheduler': False,83                 'mapping_name': 'Category',84                 'mapping_lines': ''},85                {'name': 'Sync Packages',86                 'mapping_model_name': '',87                 'python_function_name': 'sync_packages',88                 'job_type': 'system',89		 'scheduler': True,90                 'mapping_name': '',91                 'mapping_lines': ''},92                {'name': 'Sync Payments',93                 'mapping_model_name': '',94                 'python_function_name': 'sync_invoices',95                 'job_type': 'system',96		 'scheduler': True,97                 'mapping_name': '',98                 'mapping_lines': ''},99                {'name': 'Sync Configurable Relationship',100                 'mapping_model_name': 'product.template',101                 'python_function_name': 'import_configurable_links',102                 'job_type': 'system',103                 'scheduler': False,104                 'mapping_name': '',105                 'mapping_lines': ''},106                {'name': 'Sync Metadata',107                 'mapping_model_name': '',108                 'python_function_name': 'sync_mage_metadata',109                 'job_type': 'system',110                 'scheduler': False,111                 'mapping_name': '',112                 'mapping_lines': ''},113                {'name': 'Sync Attribute Data',114		 'mapping_model_name': '',115                 'python_function_name': 'sync_mage_attribute_data',116                 'job_type': 'system',117                 'scheduler': False,118                 'mapping_name': '',119                 'mapping_lines': ''},120                {'name': 'Auto Billing',121                 'mapping_model_name': '',122                 'python_function_name': 'autopay_sale_orders',123                 'job_type': 'system',124                 'scheduler': False,125                 'mapping_name': '',126                 'mapping_lines': ''},127                {'name': 'Auto Only Shipping',128                 'mapping_model_name': '',129                 'python_function_name': 'automate_only_pickings',130                 'job_type': 'system',131                 'scheduler': False,132                 'mapping_name': '',133                 'mapping_lines': ''},134                {'name': 'Auto Sale Shipping',135                 'mapping_model_name': '',136                 'python_function_name': 'autodeliver_sale_orders',137                 'job_type': 'system',138                 'scheduler': False,139                 'mapping_name': '',140                 'mapping_lines': ''},141                {'name': 'Sync Product Links',142                 'mapping_model_name': '',143                 'python_function_name': 'import_product_links',144                 'job_type': 'system',145                 'scheduler': False,146                 'mapping_name': '',147                 'mapping_lines': ''},...

lambda_functions.py

Source:lambda_functions.py

1import boto32import json3from os import path4from src.utils import *5LAMBDA_ROLE = 'Lambda_Execution_Role_new'6LAMDA_ACCESS_POLICY_ARN = 'arn:aws:iam::425439196533:policy/LambdaS3AccessPolict_new'7LAMDA_ROLE_ARN = 'arn:aws:iam::425439196533:role/Lambda_Execution_Role_new'8LAMBDA_TIMEOUT = 109LAMBDA_MEMORY = 12810PYTHON_36_RUNTIME = 'python3.6'11PYTHON_FUNCTION_NAME = 'PYTHON_LAMBDA_FUNCTION'12PYTHON_HANDLER = 'Lambda_function.handler'13def lambda_client():14    aws_lambda = boto3.client('lambda')15    return aws_lambda16def iam_client():17    return boto3.client('iam')18def create_access_policy_for_lambda():19    s3_access_policy_document = {20        "Version": "2012-10-17",21        "Statement" : [22            {23            "Action" : [24                "S3:*",25                "logs:CreateLogGroup",26                "logs:CreateLogStream",27                "logs:PutLogEvents"28            ],29            "Effect":"Allow",30            "Resource": "*"31            }32        ]33        34    }35    return iam_client().create_policy(36        PolicyName = 'LambdaS3AccessPolict_new',37        PolicyDocument = json.dumps(s3_access_policy_document),38        Description  = "Allow Lambda to access S3 resources"39        )40#'arn:aws:iam::425439196533:policy/LambdaS3AccessPolict_new'41def create_execution_role_for_lambda():42    lambda_execution_assumption = {43        "Version" : "2012-10-17",44        "Statement" : [45            {46                "Effect": "Allow",47                "Principal" : {48                    "Service" : "lambda.amazonaws.com"49                },50                "Action": "sts:AssumeRole"51            }52        ]53    }54    return iam_client().create_role(55        RoleName= LAMBDA_ROLE,56        AssumeRolePolicyDocument=json.dumps(lambda_execution_assumption),57        Description = 'Gives necessary permission to lambda'58    )59def attach_access_policy_to_execution():60    return iam_client().attach_role_policy(61        RoleName = LAMBDA_ROLE,62        PolicyArn = LAMDA_ACCESS_POLICY_ARN63    )64def deploy_lambda_function(function_name,runtime,handler,role_arn,source_folder):65    folder_path = path.join(path.dirname(path.abspath(__file__)),source_folder)66    zip_file  = Utils.make_zip_file_bytes(path = folder_path)67    return lambda_client().create_function(68        FunctionName = function_name,69        Runtime = runtime,70        Role = role_arn,71        Handler = handler,72        Code = {73            'ZipFile' : zip_file74        },75        Timeout = LAMBDA_TIMEOUT,76        MemorySize = LAMBDA_MEMORY77        #Publish = False78    )79def invoke_lambda_function(function_name):80    return lambda_client().invoke(FunctionName=function_name)81def add_environment_variable_to_lambda(function_name,variables):82    return lambda_client().update_function_configuration(83        FunctionName=function_name,84        Environment=variables85    )86if __name__=='__main__':87    #print(create_access_policy_for_lambda())88    #print(create_execution_role_for_lambda())89    #print(attach_access_policy_to_execution())90    #print(deploy_lambda_function(PYTHON_FUNCTION_NAME,PYTHON_36_RUNTIME,PYTHON_HANDLER,LAMDA_ROLE_ARN,source_folder='/Users/thimothekonchou/Datascience/Boto3_Lambda/src/Lambda'))91    #response = invoke_lambda_function(function_name=PYTHON_FUNCTION_NAME)92    #print(response['Payload'].read().decode())93    env_variables={94     "Variables" :  {95        'ENV_VAR_TEST': 'This is an environment variables'96    }97    }...

Automation Testing Tutorials

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.