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

How to use generate_service_types method in localstack

Best Python code snippet using localstack_python

scaffold.py

Source:scaffold.py

...165                return 0166        if self.is_exception:167            return 2168        return 3169def generate_service_types(output, service: ServiceModel, doc=True):170    output.write("import sys\n")171    output.write("from typing import Dict, List, Optional\n")172    output.write("from datetime import datetime\n")173    output.write("if sys.version_info >= (3, 8):\n")174    output.write("    from typing import TypedDict\n")175    output.write("else:\n")176    output.write("    from typing_extensions import TypedDict\n")177    output.write("\n")178    output.write(179        "from localstack.aws.api import handler, RequestContext, ServiceException, ServiceRequest"180    )181    output.write("\n")182    # ==================================== print type declarations183    nodes: Dict[str, ShapeNode] = {}184    for shape_name in service.shape_names:185        shape = service.shape_for(shape_name)186        nodes[shape_name] = ShapeNode(service, shape)187    # output.write("__all__ = [\n")188    # for name in nodes.keys():189    #     output.write(f'    "{name}",\n')190    # output.write("]\n")191    printed: Set[str] = set()192    visited: Set[str] = set()193    stack: List[str] = list(nodes.keys())194    stack = sorted(stack, key=lambda name: nodes[name].get_order())195    stack.reverse()196    while stack:197        name = stack.pop()198        if name in printed:199            continue200        node = nodes[name]201        dependencies = [dep for dep in node.dependencies if dep not in printed]202        if not dependencies:203            node.print_declaration(output, doc=doc)204            printed.add(name)205        elif name in visited:206            # break out of circular dependencies207            node.print_declaration(output, doc=doc, quote_types=True)208            printed.add(name)209        else:210            stack.append(name)211            stack.extend(dependencies)212            visited.add(name)213def generate_service_api(output, service: ServiceModel, doc=True):214    service_name = service.service_name.replace("-", "_")215    class_name = service_name + "_api"216    class_name = snake_to_camel_case(class_name)217    output.write(f"class {class_name}:\n")218    output.write("\n")219    output.write(f'    service = "{service.service_name}"\n')220    output.write(f'    version = "{service.api_version}"\n')221    for op_name in service.operation_names:222        operation: OperationModel = service.operation_model(op_name)223        fn_name = camel_to_snake_case(op_name)224        if operation.output_shape:225            output_shape = operation.output_shape.name226        else:227            output_shape = "None"228        output.write("\n")229        parameters = OrderedDict()230        param_shapes = OrderedDict()231        input_shape = operation.input_shape232        if input_shape is not None:233            members = list(input_shape.members)234            for m in input_shape.required_members:235                members.remove(m)236                m_shape = input_shape.members[m]237                parameters[xform_name(m)] = m_shape.name238                param_shapes[xform_name(m)] = m_shape239            for m in members:240                m_shape = input_shape.members[m]241                param_shapes[xform_name(m)] = m_shape242                parameters[xform_name(m)] = f"{m_shape.name} = None"243        if any(map(is_bad_param_name, parameters.keys())):244            # if we cannot render the parameter name, don't expand the parameters in the handler245            param_list = f"request: {input_shape.name}" if input_shape else ""246            output.write(f'    @handler("{operation.name}", expand=False)\n')247        else:248            param_list = ", ".join([f"{k}: {v}" for k, v in parameters.items()])249            output.write(f'    @handler("{operation.name}")\n')250        output.write(251            f"    def {fn_name}(self, context: RequestContext, {param_list}) -> {output_shape}:\n"252        )253        # convert html documentation to rst and print it into to the signature254        if doc:255            html = operation.documentation256            import pypandoc257            doc = pypandoc.convert_text(html, "rst", format="html")258            output.write('        """')259            output.write(f"{doc.strip()}\n")260            output.write("\n")261            # parameters262            for param_name, shape in param_shapes.items():263                # FIXME: this doesn't work properly264                pdoc = pypandoc.convert_text(shape.documentation, "rst", format="html")265                pdoc = pdoc.strip().split(".")[0] + "."266                output.write(f":param {param_name}: {pdoc}\n")267            # return value268            if operation.output_shape:269                output.write(f":returns: {operation.output_shape.name}\n")270            # errors271            for error in operation.error_shapes:272                output.write(f":raises {error.name}:\n")273            output.write('        """\n')274        output.write("        raise NotImplementedError\n")275@click.command()276@click.argument("service", type=str)277@click.option("--doc/--no-doc", default=False, help="whether or not to generate docstrings")278@click.option(279    "--save/--print",280    default=False,281    help="whether or not to save the result into the api directory",282)283def generate(service: str, doc: bool, save: bool):284    """285    Generate types and API stubs for a given AWS service.286    SERVICE is the service to generate the stubs for (e.g., sqs, or cloudformation)287    """288    from click import ClickException289    try:290        model = load_service(service)291    except UnknownServiceError:292        raise ClickException("unknown service %s" % service)293    output = io.StringIO()294    generate_service_types(output, model, doc=doc)295    generate_service_api(output, model, doc=doc)296    code = output.getvalue()297    try:298        # try to format with black299        from black import FileMode, format_str300        code = format_str(code, mode=FileMode())301    except Exception:302        pass303    if not save:304        # either just print the code to stdout305        click.echo(code)306        return307    # or find the file path and write the code to that location308    here = os.path.dirname(__file__)...

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.