Feature/assume role

README.md

@@ -26,6 +26,11 @@ For image files stored in S3, you can parse EXIF metadata including camera param
 ```bash
 uv sync --extra image
 ```
+or, if installing using ssh:
+```toml
+[dependencies]
+"S3MP[image] @ git+ssh://git@github.com/SenteraLLC/S3MP@master",
+```
 
 ```python
 from S3MP.mirror_path import MirrorPath
@@ -136,6 +141,18 @@ Although the "Setup Paths" section looks a little dense, overall this example is
 To specify the local directory to store the mirror at, use the `set_env_mirror_root` function in `global_config.py`. This will create a `.env` file in the root of the package, and will be loaded on import.
 If no mirror root is specified and no `.env` file is found, a temporary directory will be used.
 
+Alternatively, you can setup the enviroment in your project's `__init__.py` file:
+```python
+from S3MP.global_config import S3MPConfig
+
+# Set default bucket
+S3MPConfig.set_default_bucket_key("<your-default-bucket-key>")
+# Set mirror root
+S3MPConfig.set_mirror_root("s3_mirror")
+# Assume role for S3 access
+S3MPConfig.assume_role("arn:aws:iam::<account-id>:role/<role-name>")
+```
+
 ## Installation
 [uv](https://docs.astral.sh/uv/) is a fast, cross-platform Python package installer and resolver.
 

S3MP/global_config.py

@@ -5,6 +5,7 @@
 from configparser import ConfigParser
 from dataclasses import dataclass
 from pathlib import Path
+from sys import platform
 
 import boto3
 
@@ -39,12 +40,35 @@ class _S3MPConfigClass(metaclass=Singleton):
     # Config Items
     _default_bucket_key: str | None = None
     _mirror_root: Path | None = None
+    _iam_role_arn: str | None = None
 
     # Other Items
     transfer_config: S3TransferConfig | None = None
     callback: Callable | None = None
     use_async_global_thread_queue: bool = True
 
+    def assume_role(self, role_arn: str) -> None:
+        """Assume an IAM role and update the S3 client and resource with the new credentials."""
+        sts_client = boto3.client("sts")
+        assumed_role = sts_client.assume_role(
+            RoleArn=role_arn, RoleSessionName="S3MPAssumeRoleSession"
+        )
+        credentials = assumed_role["Credentials"]
+
+        self._s3_client = boto3.client(
+            "s3",
+            aws_access_key_id=credentials["AccessKeyId"],
+            aws_secret_access_key=credentials["SecretAccessKey"],
+            aws_session_token=credentials["SessionToken"],
+        )
+        self._s3_resource = boto3.resource(
+            "s3",
+            aws_access_key_id=credentials["AccessKeyId"],
+            aws_secret_access_key=credentials["SecretAccessKey"],
+            aws_session_token=credentials["SessionToken"],
+        )
+        self._iam_role_arn = role_arn
+
     @property
     def default_bucket_key(self) -> str:
         """Get default bucket key."""
@@ -93,6 +117,20 @@ def mirror_root(self) -> Path:
             self._mirror_root = Path(tempfile.gettempdir())
         return self._mirror_root
 
+    def set_mirror_root(self, mirror_root: Path | str) -> None:
+        """Set mirror root. If a relative path is provided, it will be prefixed with the OS-specific root (e.g., C:\\ on Windows or / otherwise)."""
+        mirror_path = Path(mirror_root)
+
+        if mirror_path.is_absolute():
+            # If it's an absolute path, use it as-is
+            self._mirror_root = mirror_path
+        else:
+            # If it's a relative path, prefix with OS-specific root
+            if platform == "win32":
+                self._mirror_root = Path(f"C:\\{mirror_path}")
+            else:
+                self._mirror_root = Path(f"/{mirror_path}")
+
     def load_config(self, config_file_path: Path | None = None):
         """Load the config file."""
         config_file_path = config_file_path or get_config_file_path()
@@ -108,6 +146,9 @@ def load_config(self, config_file_path: Path | None = None):
         if "mirror_root" in config["DEFAULT"]:
             self._mirror_root = Path(config["DEFAULT"]["mirror_root"])
 
+        if "iam_role_arn" in config["DEFAULT"]:
+            self.assume_role(config["DEFAULT"]["iam_role_arn"])
+
     def save_config(self, config_file_path: Path | None = None):
         """Write config file."""
         config_file_path = config_file_path or get_config_file_path()
@@ -117,6 +158,8 @@ def save_config(self, config_file_path: Path | None = None):
             config["DEFAULT"]["default_bucket_key"] = self._default_bucket_key
         if self._mirror_root:
             config["DEFAULT"]["mirror_root"] = str(self._mirror_root)
+        if self._iam_role_arn:
+            config["DEFAULT"]["iam_role_arn"] = self._iam_role_arn
         with open(config_file_path, "w") as configfile:
             config.write(configfile)
 

S3MP/utils/image_utils.py

@@ -78,6 +78,12 @@ def parse_metadata(cls, mirror_path: MirrorPath) -> ImageMetadata:
         mirror_path.download_to_mirror_if_not_present()
 
         parser = MetadataParser(mirror_path.local_path)
+
+        try:
+            distortion_params = parser.distortion_parameters()
+        except Exception:
+            distortion_params = None
+
         return cls(
             mirror_path,
             parser.dimensions(),
@@ -87,7 +93,7 @@ def parse_metadata(cls, mirror_path: MirrorPath) -> ImageMetadata:
             parser.rotation(),
             parser.focal_length_pixels(),
             parser.relative_altitude(),
-            parser.distortion_parameters(),
+            distortion_params,
         )
 
     @property

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "S3MP"
-version = "0.7.1"
+version = "0.8.0"
 description = ""
 authors = [
     {name = "Joshua Dean", email = "joshua.dean@sentera.com"},