Add get_url_path() helper to get underlying URL for a PathProxy object

Author: aron

replicate/use.py

@@ -1,10 +1,4 @@
 # TODO
-# - [x] Support downloading files and conversion into Path when schema is URL
-# - [x] Support list outputs
-# - [x] Support iterator outputs
-# - [x] Support helpers for working with ContatenateIterator
-# - [ ] Support reusing output URL when passing to new method
-# - [ ] Support lazy downloading of files into Path
 # - [ ] Support text streaming
 # - [ ] Support file streaming
 # - [ ] Support asyncio variant
@@ -28,6 +22,9 @@
 from replicate.version import Version
 
 
+__all__ = ["use", "get_path_url"]
+
+
 def _in_module_scope() -> bool:
     """
     Returns True when called from top level module scope.
@@ -41,9 +38,6 @@ def _in_module_scope() -> bool:
     return False
 
 
-__all__ = ["use"]
-
-
 def _has_concatenate_iterator_output_type(openapi_schema: dict) -> bool:
     """
     Returns true if the model output type is ConcatenateIterator or
@@ -218,29 +212,41 @@ def ensure_path() -> Path:
                 path = _download_file(target)
             return path
 
-        object.__setattr__(self, "__target__", target)
-        object.__setattr__(self, "__path__", ensure_path)
+        object.__setattr__(self, "__replicate_target__", target)
+        object.__setattr__(self, "__replicate_path__", ensure_path)
 
     def __getattribute__(self, name) -> Any:
-        if name in ("__path__", "__target__"):
+        if name in ("__replicate_path__", "__replicate_target__"):
             return object.__getattribute__(self, name)
 
         # TODO: We should cover other common properties on Path...
         if name == "__class__":
             return Path
 
-        return getattr(object.__getattribute__(self, "__path__")(), name)
+        return getattr(object.__getattribute__(self, "__replicate_path__")(), name)
 
     def __setattr__(self, name, value) -> None:
-        if name in ("__path__", "__target__"):
+        if name in ("__replicate_path__", "__replicate_target__"):
             raise ValueError()
 
-        object.__setattr__(object.__getattribute__(self, "__path__")(), name, value)
+        object.__setattr__(
+            object.__getattribute__(self, "__replicate_path__")(), name, value
+        )
 
     def __delattr__(self, name) -> None:
-        if name in ("__path__", "__target__"):
+        if name in ("__replicate_path__", "__replicate_target__"):
             raise ValueError()
-        delattr(object.__getattribute__(self, "__path__")(), name)
+        delattr(object.__getattribute__(self, "__replicate_path__")(), name)
+
+
+def get_path_url(path: Any) -> str | None:
+    """
+    Return the remote URL (if any) for a Path output from a model.
+    """
+    try:
+        return object.__getattribute__(path, "__replicate_target__")
+    except AttributeError:
+        return None
 
 
 @dataclass
@@ -252,7 +258,7 @@ class Run:
     prediction: Prediction
     schema: dict
 
-    def wait(self) -> Union[Any, Iterator[Any]]:
+    def output(self) -> Union[Any, Iterator[Any]]:
         """
         Wait for the prediction to complete and return its output.
         """
@@ -330,7 +336,7 @@ def _version(self) -> Version | None:
 
     def __call__(self, **inputs: Dict[str, Any]) -> Any:
         run = self.create(**inputs)
-        return run.wait()
+        return run.output()
 
     def create(self, **inputs: Dict[str, Any]) -> Run:
         """
@@ -341,8 +347,8 @@ def create(self, **inputs: Dict[str, Any]) -> Run:
         for key, value in inputs.items():
             if isinstance(value, OutputIterator) and value.is_concatenate:
                 processed_inputs[key] = str(value)
-            elif isinstance(value, PathProxy):
-                processed_inputs[key] = object.__getattribute__(value, "__target__")
+            elif url := get_path_url(value):
+                processed_inputs[key] = url
             else:
                 processed_inputs[key] = value
 

tests/test_use.py

@@ -608,6 +608,78 @@ async def test_use_iterator_of_paths_output(use_async_client):
     assert output_list[1].read_bytes() == b"fake image 2 data"
 
 
+def test_get_path_url_with_pathproxy():
+    """Test get_path_url returns the URL for PathProxy instances."""
+    from replicate.use import get_path_url, PathProxy
+
+    url = "https://example.com/test.jpg"
+    path_proxy = PathProxy(url)
+
+    result = get_path_url(path_proxy)
+    assert result == url
+
+
+def test_get_path_url_with_regular_path():
+    """Test get_path_url returns None for regular Path instances."""
+    from replicate.use import get_path_url
+
+    regular_path = Path("/tmp/test.txt")
+
+    result = get_path_url(regular_path)
+    assert result is None
+
+
+def test_get_path_url_with_object_without_target():
+    """Test get_path_url returns None for objects without __replicate_target__."""
+    from replicate.use import get_path_url
+
+    # Test with a string
+    result = get_path_url("not a path")
+    assert result is None
+
+    # Test with a dict
+    result = get_path_url({"key": "value"})
+    assert result is None
+
+    # Test with None
+    result = get_path_url(None)
+    assert result is None
+
+
+def test_get_path_url_with_object_with_target():
+    """Test get_path_url returns URL for any object with __replicate_target__."""
+    from replicate.use import get_path_url
+
+    class MockObjectWithTarget:
+        def __init__(self, target):
+            object.__setattr__(self, "__replicate_target__", target)
+
+    url = "https://example.com/mock.png"
+    mock_obj = MockObjectWithTarget(url)
+
+    result = get_path_url(mock_obj)
+    assert result == url
+
+
+def test_get_path_url_with_empty_target():
+    """Test get_path_url with empty/falsy target values."""
+    from replicate.use import get_path_url
+
+    class MockObjectWithEmptyTarget:
+        def __init__(self, target):
+            object.__setattr__(self, "__replicate_target__", target)
+
+    # Test with empty string
+    mock_obj = MockObjectWithEmptyTarget("")
+    result = get_path_url(mock_obj)
+    assert result == ""
+
+    # Test with None
+    mock_obj = MockObjectWithEmptyTarget(None)
+    result = get_path_url(mock_obj)
+    assert result is None
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("use_async_client", [False])
 @respx.mock