Consider AWS_ENDPOINT_URL configuration when resolving service endpoints

CHANGELOG.md

@@ -1,5 +1,6 @@
 # LocalStack Python Client Change Log
 
+* v2.1: Consider `AWS_ENDPOINT_URL` configuration when resolving service endpoints
 * v2.0: Change `LOCALSTACK_HOSTNAME` from `<hostname>` to `<hostname>:<port>`; remove `EDGE_PORT` environment variable
 * v1.39: Add endpoint for Amazon MQ
 * v1.38: Add `enable_local_endpoints()` util function; slight project refactoring, migrate from `nose` to `pytests`

README.md

@@ -70,8 +70,9 @@ assert sqs.list_queues() is not None  # list SQS in localstack
 
 You can use the following environment variables for configuration:
 
-* `LOCALSTACK_HOST`: A `<hostname>:<port>` variable defining where to find LocalStack (default: `localhost:4566`).
-* `USE_SSL`: Whether to use SSL when connecting to LocalStack (default: `False`).
+* `AWS_ENDPOINT_URL`: The endpoint URL to connect to (takes precedence over `USE_SSL`/`LOCALSTACK_HOST` below)
+* `LOCALSTACK_HOST` (deprecated): A `<hostname>:<port>` variable defining where to find LocalStack (default: `localhost:4566`).
+* `USE_SSL` (deprecated): Whether to use SSL when connecting to LocalStack (default: `False`).
 
 ### Enabling Transparent Local Endpoints
 

localstack_client/config.py

@@ -8,6 +8,7 @@
 # central entrypoint port for all LocalStack API endpoints
 DEFAULT_EDGE_PORT = 4566
 
+# TODO: deprecated, remove!
 # NOTE: The ports listed below will soon become deprecated/removed, as the default in the
 # latest version is to access all services via a single "edge service" (port 4566 by default)
 _service_ports: Dict[str, int] = {
@@ -123,21 +124,6 @@
             _service_ports[key] = DEFAULT_EDGE_PORT
 
 
-def get_service_endpoint(
-    service: str, localstack_host: Optional[str] = None
-) -> Optional[str]:
-    """
-    Return the local endpoint URL for the given boto3 service (e.g., "s3").
-    If $AWS_ENDPOINT_URL is configured in the environment, it is returned directly.
-    Otherwise, the service endpoint is constructed from the dict of service ports (usually http://localhost:4566).
-    """
-    env_endpoint_url = os.environ.get("AWS_ENDPOINT_URL", "").strip()
-    if env_endpoint_url:
-        return env_endpoint_url
-    endpoints = get_service_endpoints(localstack_host=localstack_host)
-    return endpoints.get(service)
-
-
 def parse_localstack_host(given: str) -> Tuple[str, int]:
     parts = given.split(":", 1)
     if len(parts) == 1:
@@ -156,6 +142,15 @@ def parse_localstack_host(given: str) -> Tuple[str, int]:
 
 
 def get_service_endpoints(localstack_host: Optional[str] = None) -> Dict[str, str]:
+    """
+    Return the local endpoint URLs for the list of supported boto3 services (e.g., "s3", "lambda", etc).
+    If $AWS_ENDPOINT_URL is configured in the environment, it is returned directly. Otherwise,
+    the service endpoint is constructed from the dict of service ports (usually http://localhost:4566).
+    """
+    env_endpoint_url = os.environ.get("AWS_ENDPOINT_URL", "").strip()
+    if env_endpoint_url:
+        return {key: env_endpoint_url for key in _service_ports.keys()}
+
     if localstack_host is None:
         localstack_host = os.environ.get(
             "LOCALSTACK_HOST", f"localhost:{DEFAULT_EDGE_PORT}"
@@ -168,6 +163,13 @@ def get_service_endpoints(localstack_host: Optional[str] = None) -> Dict[str, st
     return {key: f"{protocol}://{hostname}:{port}" for key in _service_ports.keys()}
 
 
+def get_service_endpoint(
+    service: str, localstack_host: Optional[str] = None
+) -> Optional[str]:
+    endpoints = get_service_endpoints(localstack_host=localstack_host)
+    return endpoints.get(service)
+
+
 def get_service_port(service: str) -> Optional[int]:
     ports = get_service_ports()
     return ports.get(service)

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = localstack-client
-version = 2.0
+version = 2.1
 url = https://github.com/localstack/localstack-python-client
 author = LocalStack Team
 author_email = info@localstack.cloud