Adding way to handle fully qualified S3 URLs

registrar/source.py

@@ -14,6 +14,7 @@ from fnmatch import fnmatch
 import logging
 from typing import TYPE_CHECKING, Optional
 from abc import ABC, abstractmethod
+from urllib.parse import urlparse
 
 import boto3
 import boto3.session
@@ -253,6 +254,14 @@ class S3Source(Source):
         )
 
     def get_container_and_path(self, path: str):
+        # try to see if we have a fully qualified S3 URL, in which case we will
+        # return bucket/path from there
+        parsed = urlparse(path)
+        if parsed.scheme:
+            if parsed.scheme.lower() != "s3":
+                raise ValueError("invalid S3 URL {path}")
+            return (parsed.netloc, parsed.path[1:])
+
         bucket = self.bucket_name
         if bucket is None:
             parts = (path[1:] if path.startswith("/") else path).split("/")