أعمل الآن على مشروع حيث يتم إنشاء REST API باستخدام AWS lambdas كمعالجات للطلب. يستخدم الأمر برمته AWS SAM لتحديد lambdas والطبقات وتوصيلها ببوابة Api في ملف template.yaml الجميل.

المشكلة

اختبار واجهة برمجة التطبيقات (API) هذه محليًا ليس سهلاً كما هو الحال مع الأطر الأخرى. بينما توفر AWS أوامر محلية sam لإنشاء صور Docker التي تستضيف lambdas (والتي تحاكي بيئة Lambda بشكل أفضل)، فقد وجدت أن هذا الأسلوب ثقيل جدًا بحيث لا يمكن التكرار السريع أثناء التطوير.

الحل

أردت طريقة لـ:

• اختبار منطق عملي والتحقق من صحة البيانات بسرعة
• توفير خادم محلي لمطوري الواجهة الأمامية لاختباره
• تجنب العبء الزائد لإعادة إنشاء صور Docker لكل تغيير

لذلك، قمت بإنشاء برنامج نصي لتلبية هذه الاحتياجات. ?‍♂️

TL;DR: قم بمراجعة server_local.py في مستودع GitHub هذا.

الفوائد الرئيسية

• الإعداد السريع: تشغيل خادم Flask محلي يقوم بتعيين مسارات بوابة API الخاصة بك إلى مسارات Flask.
• التنفيذ المباشر: يقوم بتشغيل وظيفة Python (معالج Lambda) مباشرةً، دون تحميل Docker.
• التحديث السريع: تنعكس التغييرات على الفور، مما يؤدي إلى تقصير حلقة تعليقات التطوير.

يعتمد هذا المثال على مشروع "Hello World" من sam init، مع إضافة server_local.py ومتطلباته لتمكين التطوير المحلي.

قراءة قالب SAM

ما أفعله هنا هو أنني أقرأ template.yaml أولاً نظرًا لوجود تعريف حالي للبنية التحتية الخاصة بي وجميع Lambdas.

كل الكود الذي نحتاجه لإنشاء تعريف الإملاء هو هذا. للتعامل مع الوظائف الخاصة بقالب 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"
            ]
         }
      }
   }
}

التعامل مع الطبقات

بعد أن أصبح من السهل إنشاء مسارات Flask ديناميكيًا لكل نقطة نهاية. ولكن قبل ذلك شيئا إضافيا.

في تطبيق sam init helloworld، لا توجد طبقات محددة. لكن واجهت هذه المشكلة في مشروعي الحقيقي. ولجعله يعمل بشكل صحيح، أضفت وظيفة تقرأ تعريفات الطبقات وأضفتها إلى 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)

هذا كل شيء. الكود بالكامل متاح على جيثب https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python. اسمحوا لي أن أعرف إذا وجدت أي حالة زاوية تحتوي على طبقات وما إلى ذلك. يمكن تحسين ذلك أو تعتقد أنه يستحق إضافة شيء أكثر إلى هذا. أجد أنه من المفيد جدا.

القضايا المحتملة

باختصار، يعمل هذا على بيئتك المحلية. ضع في اعتبارك أن أجهزة lambda لديها بعض القيود المطبقة على الذاكرة ووحدة المعالجة المركزية. في النهاية، من الجيد اختباره في بيئة حقيقية. يجب استخدام هذا النهج لتسريع عملية التطوير فقط.

إذا قمت بتنفيذ ذلك في مشروعك، فيرجى مشاركة أفكارك. هل عملت بشكل جيد بالنسبة لك؟ هل من تحديات واجهتك؟ تساعد تعليقاتك في تحسين هذا الحل للجميع.

هل تريد معرفة المزيد؟

ترقبوا المزيد من الأفكار والبرامج التعليمية! قم بزيارة مدونتي ?