अभी मैं एक प्रोजेक्ट पर काम कर रहा हूं जहां रिक्वेस्ट हैंडलर के रूप में AWS लैम्ब्डा का उपयोग करके REST API बनाया गया है। लैम्ब्डा, लेयर्स को परिभाषित करने और इसे अच्छे टेम्पलेट.yaml फ़ाइल में एपीआई गेटवे से जोड़ने के लिए पूरी चीज़ AWS SAM का उपयोग करती है।

समस्या

इस एपीआई का स्थानीय स्तर पर परीक्षण करना अन्य रूपरेखाओं की तरह सीधा नहीं है। जबकि AWS लैम्ब्डा को होस्ट करने वाली डॉकर छवियां बनाने के लिए सैम स्थानीय कमांड प्रदान करता है (जो लैम्ब्डा वातावरण को बेहतर ढंग से दोहराता है), मुझे विकास के दौरान त्वरित पुनरावृत्तियों के लिए यह दृष्टिकोण बहुत भारी लगा।

समाधान

मुझे एक रास्ता चाहिए था:

• मेरे व्यावसायिक तर्क और डेटा सत्यापन का त्वरित परीक्षण करें
• फ्रंटएंड डेवलपर्स के खिलाफ परीक्षण करने के लिए एक स्थानीय सर्वर प्रदान करें
• प्रत्येक परिवर्तन के लिए डॉकर छवियों के पुनर्निर्माण के ओवरहेड से बचें

इसलिए, मैंने इन जरूरतों को पूरा करने के लिए एक स्क्रिप्ट बनाई। ?‍♂️

TL;DR: इस GitHub रिपोजिटरी मेंserver_local.py देखें।

मुख्य लाभ

• त्वरित सेटअप: एक स्थानीय फ्लास्क सर्वर को स्पिन करता है जो आपके एपीआई गेटवे मार्गों को फ्लास्क मार्गों पर मैप करता है।
• प्रत्यक्ष निष्पादन: डॉकर ओवरहेड के बिना, सीधे पायथन फ़ंक्शन (लैम्ब्डा हैंडलर) को ट्रिगर करता है।
• हॉट रीलोड: परिवर्तन तुरंत प्रतिबिंबित होते हैं, जिससे विकास फीडबैक लूप छोटा हो जाता है।

यह उदाहरण सैम इनिट से "हैलो वर्ल्ड" प्रोजेक्ट पर आधारित है, जिसमें स्थानीय विकास को सक्षम करने के लिए सर्वर_local.py और इसकी आवश्यकताएं जोड़ी गई हैं।

एसएएम टेम्पलेट पढ़ना

मैं यहां जो कर रहा हूं वह यह है कि मैं सबसे पहले template.yaml पढ़ रहा हूं क्योंकि मेरे बुनियादी ढांचे और सभी लैम्ब्डा की वर्तमान परिभाषा है।

एक तानाशाही परिभाषा बनाने के लिए हमें आवश्यक सभी कोड यही हैं। SAM टेम्पलेट के लिए विशिष्ट कार्यों को संभालने के लिए मैंने CloudFormationLoader में कुछ कंस्ट्रक्टर जोड़े हैं। अब यह किसी अन्य ऑब्जेक्ट के संदर्भ के रूप में Ref, विकल्प के लिए विधि के रूप में Sub और विशेषताएँ प्राप्त करने के लिए GetAtt का समर्थन कर सकता है। मुझे लगता है कि हम यहां और अधिक तर्क जोड़ सकते हैं लेकिन अभी इसे काम करने के लिए यह पूरी तरह से पर्याप्त था।

import os
from typing import Any, Dict
import yaml


class CloudFormationLoader(yaml.SafeLoader):
    def __init__(self, stream):
        self._root = os.path.split(stream.name)[0]  # type: ignore
        super(CloudFormationLoader, self).__init__(stream)

    def include(self, node):
        filename = os.path.join(self._root, self.construct_scalar(node))  # type: ignore
        with open(filename, "r") as f:
            return yaml.load(f, CloudFormationLoader)


def construct_getatt(loader, node):
    if isinstance(node, yaml.ScalarNode):
        return {"Fn::GetAtt": loader.construct_scalar(node).split(".")}
    elif isinstance(node, yaml.SequenceNode):
        return {"Fn::GetAtt": loader.construct_sequence(node)}
    else:
        raise yaml.constructor.ConstructorError(
            None, None, f"Unexpected node type for !GetAtt: {type(node)}", node.start_mark
        )


CloudFormationLoader.add_constructor(
    "!Ref", lambda loader, node: {"Ref": loader.construct_scalar(node)}  # type: ignore
)
CloudFormationLoader.add_constructor(
    "!Sub", lambda loader, node: {"Fn::Sub": loader.construct_scalar(node)}  # type: ignore
)
CloudFormationLoader.add_constructor("!GetAtt", construct_getatt)


def load_template() -> Dict[str, Any]:
    with open("template.yaml", "r") as file:
        return yaml.load(file, Loader=CloudFormationLoader)

और यह इस तरह json उत्पन्न करता है:

{
   "AWSTemplateFormatVersion":"2010-09-09",
   "Transform":"AWS::Serverless-2016-10-31",
   "Description":"sam-app\nSample SAM Template for sam-app\n",
   "Globals":{
      "Function":{
         "Timeout":3,
         "MemorySize":128,
         "LoggingConfig":{
            "LogFormat":"JSON"
         }
      }
   },
   "Resources":{
      "HelloWorldFunction":{
         "Type":"AWS::Serverless::Function",
         "Properties":{
            "CodeUri":"hello_world/",
            "Handler":"app.lambda_handler",
            "Runtime":"python3.9",
            "Architectures":[
               "x86_64"
            ],
            "Events":{
               "HelloWorld":{
                  "Type":"Api",
                  "Properties":{
                     "Path":"/hello",
                     "Method":"get"
                  }
               }
            }
         }
      }
   },
   "Outputs":{
      "HelloWorldApi":{
         "Description":"API Gateway endpoint URL for Prod stage for Hello World function",
         "Value":{
            "Fn::Sub":"https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/hello/"
         }
      },
      "HelloWorldFunction":{
         "Description":"Hello World Lambda Function ARN",
         "Value":{
            "Fn::GetAtt":[
               "HelloWorldFunction",
               "Arn"
            ]
         }
      },
      "HelloWorldFunctionIamRole":{
         "Description":"Implicit IAM Role created for Hello World function",
         "Value":{
            "Fn::GetAtt":[
               "HelloWorldFunctionRole",
               "Arn"
            ]
         }
      }
   }
}

परतों को संभालना

ऐसा होने से प्रत्येक समापन बिंदु के लिए गतिशील रूप से फ्लास्क मार्ग बनाना आसान हो जाता है। लेकिन उससे पहले कुछ अतिरिक्त.

सैम इनिट हेलोवर्ल्ड ऐप में कोई परतें परिभाषित नहीं हैं। लेकिन मेरे वास्तविक प्रोजेक्ट में यह समस्या थी। इसे ठीक से काम करने के लिए मैंने एक फ़ंक्शन जोड़ा है जो परतों की परिभाषाओं को पढ़ता है और उन्हें sys.path में जोड़ता है ताकि पायथन आयात सही ढंग से काम कर सके। इसे देखो:

def add_layers_to_path(template: Dict[str, Any]):
    """Add layers to path. Reads the template and adds the layers to the path for easier imports."""
    resources = template.get("Resources", {})
    for _, resource in resources.items():
        if resource.get("Type") == "AWS::Serverless::LayerVersion":
            layer_path = resource.get("Properties", {}).get("ContentUri")
            if layer_path:
                full_path = os.path.join(os.getcwd(), layer_path)
                if full_path not in sys.path:
                    sys.path.append(full_path)

फ्लास्क रूट बनाना

हमें पूरे संसाधनों में लूप करने और सभी फ़ंक्शन ढूंढने की आवश्यकता है। उसके आधार पर मैं फ्लास्क मार्गों के लिए डेटा की आवश्यकता बना रहा हूं।

def export_endpoints(template: Dict[str, Any]) -> List[Dict[str, str]]:
    endpoints = []
    resources = template.get("Resources", {})
    for resource_name, resource in resources.items():
        if resource.get("Type") == "AWS::Serverless::Function":
            properties = resource.get("Properties", {})
            events = properties.get("Events", {})
            for event_name, event in events.items():
                if event.get("Type") == "Api":
                    api_props = event.get("Properties", {})
                    path = api_props.get("Path")
                    method = api_props.get("Method")
                    handler = properties.get("Handler")
                    code_uri = properties.get("CodeUri")

                    if path and method and handler and code_uri:
                        endpoints.append(
                            {
                                "path": path,
                                "method": method,
                                "handler": handler,
                                "code_uri": code_uri,
                                "resource_name": resource_name,
                            }
                        )
    return endpoints

फिर अगला चरण इसका उपयोग करना और प्रत्येक के लिए एक मार्ग स्थापित करना है।

def setup_routes(template: Dict[str, Any]):
    endpoints = export_endpoints(template)
    for endpoint in endpoints:
        setup_route(
            endpoint["path"],
            endpoint["method"],
            endpoint["handler"],
            endpoint["code_uri"],
            endpoint["resource_name"],
        )


def setup_route(path: str, method: str, handler: str, code_uri: str, resource_name: str):
    module_name, function_name = handler.rsplit(".", 1)
    module_path = os.path.join(code_uri, f"{module_name}.py")
    spec = importlib.util.spec_from_file_location(module_name, module_path)
    if spec is None or spec.loader is None:
        raise Exception(f"Module {module_name} not found in {code_uri}")
    module = importlib.util.module_from_spec(spec)

    spec.loader.exec_module(module)
    handler_function = getattr(module, function_name)

    path = path.replace("{", "")

    print(f"Setting up route for [{method}] {path} with handler {resource_name}.")

    # Create a unique route handler for each Lambda function
    def create_route_handler(handler_func):
        def route_handler(*args, **kwargs):
            event = {
                "httpMethod": request.method,
                "path": request.path,
                "queryStringParameters": request.args.to_dict(),
                "headers": dict(request.headers),
                "body": request.get_data(as_text=True),
                "pathParameters": kwargs,
            }
            context = LambdaContext(resource_name)
            response = handler_func(event, context)

            try:
                api_response = APIResponse(**response)
                headers = response.get("headers", {})
                return Response(
                    api_response.body,
                    status=api_response.statusCode,
                    headers=headers,
                    mimetype="application/json",
                )
            except ValidationError as e:
                return jsonify({"error": "Invalid response format", "details": e.errors()}), 500

        return route_handler

    # Use a unique endpoint name for each route
    endpoint_name = f"{resource_name}_{method}_{path.replace('/', '_')}"
    app.add_url_rule(
        path,
        endpoint=endpoint_name,
        view_func=create_route_handler(handler_function),
        methods=[method.upper(), "OPTIONS"],
    )

और आप अपना सर्वर शुरू कर सकते हैं

if __name__ == "__main__":
    template = load_template()
    add_layers_to_path(template)
    setup_routes(template)
    app.run(debug=True, port=3000)

इतना ही। संपूर्ण कोड github https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python पर उपलब्ध है। यदि आपको परतों आदि वाला कोई कोने वाला केस मिले तो मुझे बताएं। उसमें सुधार किया जा सकता है या आप सोचते हैं कि इसमें कुछ और जोड़ना उचित है। मुझे यह बहुत मददगार लगता है.

संभावित मुद्दे

संक्षेप में यह आपके स्थानीय परिवेश पर काम करता है। ध्यान रखें कि लैम्ब्डा में कुछ मेमोरी सीमाएँ लागू होती हैं और सीपीयू। अंत में वास्तविक वातावरण में इसका परीक्षण करना अच्छा है। विकास प्रक्रिया को गति देने के लिए इस दृष्टिकोण का उपयोग किया जाना चाहिए।

यदि आप इसे अपने प्रोजेक्ट में लागू करते हैं, तो कृपया अपनी अंतर्दृष्टि साझा करें। क्या इसने आपके लिए अच्छा काम किया? क्या आपको किसी चुनौती का सामना करना पड़ा? आपकी प्रतिक्रिया सभी के लिए इस समाधान को बेहतर बनाने में मदद करती है।

और अधिक जानने की इच्छा है?

अधिक जानकारी और ट्यूटोरियल के लिए बने रहें! मेरे ब्लॉग पर जाएँ?