localstack · viren-nadkarni · Nov 24, 2022 · Jan 2, 2023 · Jan 4, 2023 · Jan 4, 2023
diff --git a/localstack/aws/client.py b/localstack/aws/client.py
@@ -1,29 +1,17 @@
 """Utils to process AWS requests as a client."""
-import dataclasses
 import io
-import json
 import logging
    from datetime import datetime
-from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, TypedDict
+from typing import Dict, Iterable, Optional
 
-from boto3 import Session
-from botocore.awsrequest import AWSPreparedRequest
-from botocore.client import BaseClient
-from botocore.config import Config as BotoConfig
 from botocore.model import OperationModel
 from botocore.parsers import ResponseParser, ResponseParserFactory
 from werkzeug import Response
 
 from localstack.aws.api import CommonServiceException, ServiceException, ServiceResponse
-from localstack.constants import INTERNAL_AWS_ACCESS_KEY_ID, INTERNAL_AWS_SECRET_ACCESS_KEY
 from localstack.runtime import hooks
-from localstack.utils.aws.arns import extract_region_from_arn
-from localstack.utils.aws.aws_stack import get_local_service_url
 from localstack.utils.patch import patch
 
-if TYPE_CHECKING:
-    from mypy_boto3_sqs import SQSClient
-
 LOG = logging.getLogger(__name__)
 
 
@@ -232,208 +220,3 @@ def raise_service_exception(response: Response, parsed_response: Dict) -> None:
     """
     if service_exception := parse_service_exception(response, parsed_response):
         raise service_exception
-
-
-#
-# Internal AWS client
-#
-
-"""
-The internal AWS client API provides the means to perform cross-service
-communication within LocalStack.
-
-Any additional information LocalStack might need for the purpose of policy
-enforcement is sent as a data transfer object. This is a serialised dict object
-sent in the request header.
-"""
-
-LOCALSTACK_DATA_HEADER = "x-localstack-data"
-"""Request header which contains the data transfer object."""
-
-
-class LocalStackData(TypedDict):
-    """
-    LocalStack Data Transfer Object.
-    """
-
-    source_arn: str
-    source_service: str  # eg. 'ec2.amazonaws.com'
-
-
-class Credentials(TypedDict):
-    """
-    AWS credentials.
-    """
-
-    aws_access_key_id: str
-    aws_secret_access_key: str
-    aws_session_token: str
-
-
-@dataclasses.dataclass(frozen=True)
-class ClientOptions:
-    """This object holds configuration options for the internal AWS client."""
-
-    region_name: Optional[str] = None
-    """Name of the AWS region to be associated with the client."""
-
-    endpoint_url: Optional[str] = None
-    """Full endpoint URL to be used by the client."""
-
-    use_ssl: bool = True
-    """Whether or not to use SSL."""
-
-    verify: bool = True
-    """Whether or not to verify SSL certificates."""
-
-    aws_access_key_id: Optional[str] = None
-    """Access key to use for the client."""
-
-    aws_secret_access_key: Optional[str] = None
-    """Secret key to use for the client."""
-
-    aws_session_token: Optional[str] = None
-    """Session token to use for the client"""
-
-    boto_config: Optional[BotoConfig] = dataclasses.field(default_factory=BotoConfig)
-    """Boto client configuration for advanced use."""
-
-    localstack_data: dict[str, Any] = dataclasses.field(default_factory=LocalStackData)
-    """LocalStack data transfer object."""
-
-
-class ClientFactory:
-    """
-    Factory to build the internal AWS client.
-    """
-
-    # TODO migrate to immutable clientfactory instances
-    client_options: ClientOptions
-    session: Session
-
-    def __init__(self, client_options: ClientOptions = None):
-        self.client_options = client_options or ClientOptions()
-        self.session = Session()
-
-    def with_endpoint(self, endpoint: str) -> "ClientFactory":
-        """
-        Set a custom endpoint.
-        """
-        return ClientFactory(
-            client_options=dataclasses.replace(self.client_options, endpoint_url=endpoint)
-        )
-
-    def with_source_arn(self, arn: str) -> "ClientFactory":
-        """
-        Indicate that the client is operating from a given resource.
-
-        This must be used in cross-service requests.
-        """
-        return ClientFactory(
-            client_options=dataclasses.replace(
-                self.client_options,
-                localstack_data=self.client_options.localstack_data
-                | LocalStackData(source_arn=arn),
-            )
-        )
-
-    def with_target_arn(self, arn: str) -> "ClientFactory":
-        """
-        Create the client to operate on a target resource.
-
-        This must be used in cross-service requests.
-        """
-        region_name = extract_region_from_arn(arn)
-        return ClientFactory(
-            client_options=dataclasses.replace(self.client_options, region_name=region_name)
-        )
-
-    def with_source_service_principal(self, source_service: str) -> "ClientFactory":
-        """
-        Set the source service principal.
-
-        This must be used in cross-service requests.
-        """
-        return ClientFactory(
-            client_options=dataclasses.replace(
-                self.client_options,
-                localstack_data=self.client_options.localstack_data
-                | LocalStackData(source_service=f"{source_service}.amazonaws.com"),
-            )
-        )
-
-    def with_credentials(
-        self, aws_access_key_id: str, aws_secret_access_key: str
-    ) -> "ClientFactory":
-        """
-        Use custom AWS credentials.
-        """
-        return ClientFactory(
-            client_options=dataclasses.replace(
-                self.client_options,
-                aws_access_key_id=aws_access_key_id,
-                aws_secret_access_key=aws_secret_access_key,
-            )
-        )
-
-    def with_env_credentials(self) -> "ClientFactory":
-        """
-        Use AWS credentials from the following locations:
-        - Environment variables
-        - Credentials file `~/.aws/credentials`
-        - Config file `~/.aws/config`
-        """
-        credentials = self.session.get_credentials()
-        return self.with_credentials(credentials.access_key, credentials.secret_key)
-
-    def with_boto_config(self, config: BotoConfig) -> "ClientFactory":
-        """
-        Use a custom BotoConfig.
-        """
-        return ClientFactory(
-            client_options=dataclasses.replace(
-                self.client_options, boto_config=self.client_options.boto_config.merge(config)
-            )
-        )
-
-    def build(self, service: str) -> BaseClient:
-        """
-        Finalise the client.
-        """
-        aws_access_key_id = self.client_options.aws_access_key_id or INTERNAL_AWS_ACCESS_KEY_ID
-        aws_secret_access_key = (
-            self.client_options.aws_secret_access_key or INTERNAL_AWS_SECRET_ACCESS_KEY
-        )
-        endpoint_url = self.client_options.endpoint_url or get_local_service_url(service)
-
-        # TODO@viren: creating a boto client is very intensive. In old aws_stack, we cache clients based on
-        # [service_name, client, env, region, endpoint_url, config, internal, kwargs]
-        # Come up with an appropriate solution here
-        client = self.session.client(
-            service_name=service,
-            config=self.client_options.boto_config,
-            aws_access_key_id=aws_access_key_id,
-            aws_secret_access_key=aws_secret_access_key,
-            endpoint_url=endpoint_url,
-        )
-
-        def event_handler(request: AWSPreparedRequest, **_):
-            # Send a compact JSON representation as DTO
-            request.headers[LOCALSTACK_DATA_HEADER] = json.dumps(
-                self.client_options.localstack_data, separators=(",", ":")
-            )
-
-        client.meta.events.register("before-send.*.*", handler=event_handler)
-
-        return client
-
-    #
-    # Convenience helpers
-    #
-
-    def sqs(self) -> "SQSClient":
-        return self.build("sqs")
-
-
-def aws_client():
-    return ClientFactory()