Merge pull request #24 from philvarner/pv/endpoint_url

endpoint url

CHANGELOG.md

@@ -6,6 +6,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Added
+
+- boto3utils.s3 object now supports specifying the `endpoint_url` parameter
+  for the underlying boto3 client
+
 ## [v0.4.1] - 2022-11-19
 
 ### Fixed

boto3utils/s3.py

@@ -5,7 +5,7 @@
 import logging
 import os
 import os.path as op
-from typing import Tuple
+from typing import Tuple, Optional
 
 from botocore.exceptions import ClientError
 from copy import deepcopy
@@ -21,12 +21,17 @@
 
 
 class s3(object):
-    def __init__(self, session=None, requester_pays=False):
+    def __init__(
+        self,
+        session: boto3.Session = None,
+        requester_pays: bool = False,
+        endpoint_url: Optional[str] = None,
+    ):
         self.requester_pays = requester_pays
         if session is None:
-            self.s3 = boto3.client("s3")
+            self.s3 = boto3.client("s3", endpoint_url=endpoint_url)
         else:
-            self.s3 = session.client("s3")
+            self.s3 = session.client("s3", endpoint_url=endpoint_url)
 
     @classmethod
     def urlparse(cls, url):

tests/conftest.py

@@ -31,6 +31,13 @@ def s3_west(aws_credentials):
         yield boto3.client("s3", region_name="us-west-2")
 
 
+@pytest.fixture
+def s3_custom_endpoint(aws_credentials):
+    os.environ["MOTO_S3_CUSTOM_ENDPOINTS"] = "http://my-s3"
+    with moto.mock_s3():
+        yield boto3.client("s3", endpoint_url="http://my-s3")
+
+
 @pytest.fixture
 def secretsmanager(aws_credentials):
     with moto.mock_secretsmanager():

tests/test_s3.py

@@ -188,3 +188,10 @@ def test_latest_inventory():
     #    hours = (datetime.today() - dt).seconds // 3600
     #    assert(hours < 24)
     #    break
+
+
+def test_upload_download_with_custom_endpoint(s3_custom_endpoint):
+    create_test_bucket(s3_custom_endpoint, BUCKET)
+    url = "s3://%s/mytestfile" % BUCKET
+    s3(endpoint_url="http://my-s3").upload(__file__, url, public=True)
+    assert s3(endpoint_url="http://my-s3").exists(url)