KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use cognito_user_pool_arn method in localstack

Best Python code snippet using localstack_python

route.py

Source:route.py

1import logging2import os3from pathlib import Path4from runpy import run_path5from typing import Any, Optional, Sequence6import pulumi_aws as aws7import pulumi_aws_apigateway as api8from openapi_schema_pydantic import OpenAPI9from openapi_schema_pydantic.util import construct_open_api_with_schema_class10from pulumi import ComponentResource, Output, ResourceOptions, get_stack11from pulumi_aws.iam import RolePolicy12from recover.const import ACCEPTED_API_METHODS, API_AUTHORIZATION_HEADER13from recover.iac._lambda import Lambda, LambdaArgs14logger = logging.getLogger(__name__)15class RouteFactory:16    @staticmethod17    def _get_endpoint(path: Path, lambda_resource_path: Path) -> str:18        """19        extracts the endpoint name from relational parental path:20        example:21            for given path = ../../src/sl/products/delivery/post.py22            the endpoint is: /products/delivery23            this endpoint is used to call the api using24            specified method in the handler python file(i.e. POST method for post.py)25        """26        endpoint = str(path.parent).replace(str(lambda_resource_path), "")27        return endpoint if endpoint[0] == "/" else ("/" + endpoint)28    @staticmethod29    def _get_request_params(path: Path):30        validation = run_path(str(path))31        request_params = validation.get("request_params")32        if not request_params:33            return []34        return [api.RequiredParameterArgs(**params) for params in request_params]35    @staticmethod36    def _get_authorizers(cognito_user_pool_arn) -> Optional[Sequence[api.AuthorizerArgs]]:37        authorizers = []38        if get_stack() != "localstack":39            if cognito_user_pool_arn is not None:40                authorizers.append(41                    api.AuthorizerArgs(42                        parameter_name="Authorization",43                        identity_source=[API_AUTHORIZATION_HEADER],44                        provider_arns=[cognito_user_pool_arn],45                    )46                )47        return authorizers48    @staticmethod49    def _get_base_docs(app_name: str, app_version: str, api_url: str) -> dict[str, Any]:50        return {51            "info": {"title": f"{app_name} API", "version": app_version},52            "servers": [{"description": "base url", "url": api_url}],53            "paths": {},54        }55    @staticmethod56    def _add_schema(docs_dict: dict[str, Any], rest_method: str, endpoint: str, path: Path) -> None:57        """Adds the schema found in the method file to the openapi specs."""58        validation = run_path(str(path))59        schema = validation.get("schema")60        if schema is None:61            raise Exception(62                f"schema is missing in {path}. "63                f"Every API method requires a schema for validation and documentation, please add one."64            )65        if schema == "ignore":66            return67        if endpoint in docs_dict["paths"]:68            docs_dict["paths"][endpoint][rest_method] = schema69        else:70            docs_dict["paths"][endpoint] = {rest_method: schema}71    @staticmethod72    def _generate_openapi(docs_dict: dict[str, Any], lambda_resource_path: Path) -> None:73        open_api = construct_open_api_with_schema_class(OpenAPI.parse_obj(docs_dict))74        # Write the resulting openapi specs to a standard location within the api75        openapi_file_path = lambda_resource_path.joinpath("docs/openapi.json")76        openapi_file_path.unlink(missing_ok=True)77        with open(openapi_file_path, "w") as f:78            f.write(open_api.json(by_alias=True, exclude_none=True, indent=2))79    @staticmethod80    def build(81        app_name: str,82        app_version: str,83        base_api_path: Path,84        lambda_resource_path: Path,85        default_role_arn: Output[str],86        default_policy: RolePolicy,87        cognito_user_pool_arn: Output[str],88        default_lambda_layers: Optional[list[aws.lambda_.LayerVersion]] = None,89        parent_api: Optional[ComponentResource] = None,90    ):91        docs_dict = RouteFactory._get_base_docs(app_name=app_name, app_version=app_version, api_url=str(base_api_path))92        routes = []93        for current_path, folders, files in os.walk(lambda_resource_path):94            """95            For example:96            file:  get.py97            lambda_resource_path:  /source/api/v198            rest_method:  get99            endpoint:  /subscriber100            """101            for file in files:102                path = Path(os.path.join(current_path, file))103                rest_method = path.stem104                if rest_method not in ACCEPTED_API_METHODS:105                    continue106                endpoint = RouteFactory._get_endpoint(path=path, lambda_resource_path=lambda_resource_path)107                resource_name = f"{app_name}-lambda{'-'.join(endpoint.split('/'))}-{rest_method}"108                event_handler = Lambda(109                    resource_name,110                    args=LambdaArgs(111                        method=rest_method,112                        endpoint=endpoint,113                        func_dir=f"{lambda_resource_path}{endpoint}",114                        func_handler=f"{rest_method}.handler",115                        func_name=resource_name,116                        default_role_arn=default_role_arn,117                        default_lambda_layers=default_lambda_layers,118                    ),119                    opts=ResourceOptions(parent=parent_api, depends_on=[default_policy]),120                ).lambda_function121                routes.append(122                    api.RouteArgs(123                        path=f"{base_api_path}{endpoint}",124                        method=api.Method(rest_method.upper()),125                        event_handler=event_handler,126                        # Define a request authorizer which uses Lambda or cognito user pool to validate the token127                        # from the Authorization header128                        authorizers=RouteFactory._get_authorizers(cognito_user_pool_arn),129                        api_key_required=False,130                    )131                )132                # Add the endpoint schema to the openapi docs133                RouteFactory._add_schema(docs_dict=docs_dict, rest_method=rest_method, endpoint=endpoint, path=path)134        # Finally generate the openapi specs and add local static files to the routes135        RouteFactory._generate_openapi(docs_dict=docs_dict, lambda_resource_path=lambda_resource_path)136        routes.append(api.RouteArgs(path="/", local_path=str(lambda_resource_path / "docs")))...

Source:register.py

1import os2from boto3 import resource, client3from chalice import Chalice, CognitoUserPoolAuthorizer4from chalicelib.dependency.cognito import CognitoUsername, FakeUsername5from chalicelib.dependency.injection import _Dependencies6APP_NAME = os.environ.get('APP_NAME', '')7USER_TABLE_NAME = os.environ.get('USER_TABLE_NAME', '')8MEDIA_BUCKET_NAME = os.environ.get('MEDIA_BUCKET_NAME', '')9COGNITO_USER_POOL_ARN = os.environ.get('COGNITO_USER_POOL_ARN', '')10_dependencies = _Dependencies()11class DependencyRegister:12    """Place to register app dependencies"""13    _instance = None14    __container = {}15    app: Chalice16    authorizer: CognitoUserPoolAuthorizer17    def __new__(cls):18        if cls._instance is None:19            cls._instance = super().__new__(cls)20        return cls._instance21    def __init__(self):22        self.__add_app()23        self.__add_database()24        self.__add_storage()25        self.__add_authorizer()26    def __getattr__(self, dependency_name):27        return self.__container.get(dependency_name)28    def __add_app(self):29        self.__app = Chalice(app_name=APP_NAME)30        self.__container['app'] = self.__app31    def __add_database(self):32        dynamodb = resource('dynamodb')33        self.__container['user_table'] = dynamodb.Table(USER_TABLE_NAME)34    def __add_storage(self):35        self.__container['s3'] = client('s3')36        self.__container['media_bucket_name'] = MEDIA_BUCKET_NAME37    def __add_authorizer(self):38        if COGNITO_USER_POOL_ARN:39            _dependencies.register(CognitoUsername(app=self.__app))40            self.__container['authorizer'] = CognitoUserPoolAuthorizer(41                'CognitoAuthorizer',42                provider_arns=[COGNITO_USER_POOL_ARN],43                header='Authorization', scopes=None44            )45        else:46            _dependencies.register(FakeUsername())...

handler.py

Source:handler.py

1import os2import re3import json4import boto35COGNITO_USER_POOL_ARN = os.environ.get('COGNITO_USER_POOL_ARN')6COGNITO_IDENTITY_POOL_ID = os.environ.get('COGNITO_IDENTITY_POOL_ID')7def _get_identity_id_from_cognito_idToken(idToken):8    pattern = 'arn:aws:cognito-idp:(?P<aws_region>.*):(?P<aws_account_id>[0-9]{12}):userpool/(?P<user_pool_id>.*)'9    m = re.match(pattern, COGNITO_USER_POOL_ARN)10    region = m.group('aws_region')11    account_id = m.group('aws_account_id')12    user_pool_id = m.group('user_pool_id')13    cognito_user_pool_keyname = f"cognito-idp.{region}.amazonaws.com/{user_pool_id}"14    # @see https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cognito-identity.html#CognitoIdentity.Client.get_id15    response = boto3.client('cognito-identity', region).get_id(16        AccountId=account_id,17        IdentityPoolId=COGNITO_IDENTITY_POOL_ID,18        Logins={19            # Amazon Cognito user pool Example:20            #   "cognito-idp.us-east-1.amazonaws.com/us-east-1_123456789": "${idToken}"21            cognito_user_pool_keyname: idToken,22        }23    )24    return response['IdentityId']25def hello(event, context):26    idToken = event['headers']['Authorization']27    body = {28        "message": "Go Serverless v1.0! Your function executed successfully!",29        "input": event,30        "identity_id": _get_identity_id_from_cognito_idToken(idToken)31    }32    response = {33        # CORS settings for Lambda Proxy Integration34        "headers": {35            "Access-Control-Allow-Origin": '*'36        },37        "statusCode": 200,38        "body": json.dumps(body)39    }...

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.