feat: add backward compatibility for models.get("owner/name") syntax

This PR adds backward compatibility for the legacy models.get("owner/name")
syntax while maintaining full forward compatibility with the new keyword
argument format.

- Add compatibility layer in lib/models.py that handles both formats
- Patch both sync and async ModelsResource instances in client initialization
- Support both models.get("stability-ai/stable-diffusion") and
  models.get(model_owner="stability-ai", model_name="stable-diffusion")
- Add comprehensive tests for both syntax formats and error cases
- Reduce breaking changes from 4 to 3 areas for easier migration

Resolves Linear issue DP-656

Author: zeke

src/replicate/_client.py

@@ -186,9 +186,11 @@ def account(self) -> AccountResource:
 
     @cached_property
     def models(self) -> ModelsResource:
+        from .lib.models import patch_models_resource
         from .resources.models import ModelsResource
 
-        return ModelsResource(self)
+        models_resource = ModelsResource(self)
+        return patch_models_resource(models_resource)
 
     @cached_property
     def predictions(self) -> PredictionsResource:
@@ -572,9 +574,11 @@ def account(self) -> AsyncAccountResource:
 
     @cached_property
     def models(self) -> AsyncModelsResource:
+        from .lib.models import patch_models_resource
         from .resources.models import AsyncModelsResource
 
-        return AsyncModelsResource(self)
+        models_resource = AsyncModelsResource(self)
+        return patch_models_resource(models_resource)
 
     @cached_property
     def predictions(self) -> AsyncPredictionsResource:

src/replicate/lib/models.py

@@ -0,0 +1,119 @@
+"""
+Custom models functionality with backward compatibility.
+"""
+
+from __future__ import annotations
+
+import inspect
+from typing import TYPE_CHECKING, Union
+
+from .._types import NOT_GIVEN, NotGiven
+from ._models import ModelVersionIdentifier
+
+if TYPE_CHECKING:
+    import httpx
+
+    from .._types import Body, Query, Headers
+    from ..resources.models.models import ModelsResource, AsyncModelsResource
+    from ..types.model_get_response import ModelGetResponse
+
+
+def _parse_model_args(
+    model_or_owner: str | NotGiven,
+    model_owner: str | NotGiven,
+    model_name: str | NotGiven,
+) -> tuple[str, str]:
+    """Parse model arguments and return (owner, name)."""
+    # Handle legacy format: models.get("owner/name")
+    if model_or_owner is not NOT_GIVEN:
+        if model_owner is not NOT_GIVEN or model_name is not NOT_GIVEN:
+            raise ValueError(
+                "Cannot specify both positional 'model_or_owner' and keyword arguments "
+                "'model_owner'/'model_name'. Use either the legacy format "
+                "models.get('owner/name') or the new format models.get(model_owner='owner', model_name='name')."
+            )
+
+        # Type guard: model_or_owner is definitely a string here
+        assert isinstance(model_or_owner, str)
+
+        # Parse the owner/name format
+        if "/" not in model_or_owner:
+            raise ValueError(
+                f"Invalid model reference '{model_or_owner}'. "
+                "Expected format: 'owner/name' (e.g., 'stability-ai/stable-diffusion')"
+            )
+
+        try:
+            parsed = ModelVersionIdentifier.parse(model_or_owner)
+            return parsed.owner, parsed.name
+        except ValueError as e:
+            raise ValueError(
+                f"Invalid model reference '{model_or_owner}'. "
+                f"Expected format: 'owner/name' (e.g., 'stability-ai/stable-diffusion'). "
+                f"Error: {e}"
+            ) from e
+
+    # Validate required parameters for new format
+    if model_owner is NOT_GIVEN or model_name is NOT_GIVEN:
+        raise ValueError(
+            "model_owner and model_name are required. "
+            "Use either models.get('owner/name') or models.get(model_owner='owner', model_name='name')"
+        )
+
+    return model_owner, model_name
+
+
+def patch_models_resource(
+    models_resource: Union["ModelsResource", "AsyncModelsResource"],
+) -> Union["ModelsResource", "AsyncModelsResource"]:
+    """Patch a models resource to add backward compatibility."""
+    original_get = models_resource.get
+    is_async = inspect.iscoroutinefunction(original_get)
+
+    if is_async:
+
+        async def get_wrapper(
+            model_or_owner: str | NotGiven = NOT_GIVEN,
+            *,
+            model_owner: str | NotGiven = NOT_GIVEN,
+            model_name: str | NotGiven = NOT_GIVEN,
+            extra_headers: "Headers | None" = None,
+            extra_query: "Query | None" = None,
+            extra_body: "Body | None" = None,
+            timeout: "float | httpx.Timeout | None | NotGiven" = NOT_GIVEN,
+        ) -> "ModelGetResponse":
+            owner, name = _parse_model_args(model_or_owner, model_owner, model_name)
+            return await original_get(
+                model_owner=owner,
+                model_name=name,
+                extra_headers=extra_headers,
+                extra_query=extra_query,
+                extra_body=extra_body,
+                timeout=timeout,
+            )
+    else:
+
+        def get_wrapper(
+            model_or_owner: str | NotGiven = NOT_GIVEN,
+            *,
+            model_owner: str | NotGiven = NOT_GIVEN,
+            model_name: str | NotGiven = NOT_GIVEN,
+            extra_headers: "Headers | None" = None,
+            extra_query: "Query | None" = None,
+            extra_body: "Body | None" = None,
+            timeout: "float | httpx.Timeout | None | NotGiven" = NOT_GIVEN,
+        ) -> "ModelGetResponse":
+            owner, name = _parse_model_args(model_or_owner, model_owner, model_name)
+            return original_get(
+                model_owner=owner,
+                model_name=name,
+                extra_headers=extra_headers,
+                extra_query=extra_query,
+                extra_body=extra_body,
+                timeout=timeout,
+            )
+
+    # Store original method for tests and replace with wrapper
+    models_resource._original_get = original_get
+    models_resource.get = get_wrapper
+    return models_resource

tests/test_models_backward_compat.py

@@ -0,0 +1,238 @@
+"""
+Tests for backward compatibility in models.get() method.
+"""
+
+from unittest.mock import Mock, AsyncMock, patch
+
+import pytest
+
+from replicate import Replicate, AsyncReplicate
+from replicate._types import NOT_GIVEN
+from replicate.types.model_get_response import ModelGetResponse
+
+
+@pytest.fixture
+def mock_model_response():
+    """Mock response for model.get requests."""
+    return ModelGetResponse(
+        url="https://replicate.com/stability-ai/stable-diffusion",
+        owner="stability-ai",
+        name="stable-diffusion",
+        description="A model for generating images from text prompts",
+        visibility="public",
+        github_url=None,
+        paper_url=None,
+        license_url=None,
+        run_count=0,
+        cover_image_url=None,
+        default_example=None,
+        latest_version=None,
+    )
+
+
+class TestModelGetBackwardCompatibility:
+    """Test backward compatibility for models.get() method."""
+
+    @pytest.fixture
+    def client(self):
+        """Create a Replicate client with mocked token."""
+        with patch("replicate.lib.cog._get_api_token_from_environment", return_value="test-token"):
+            return Replicate()
+
+    def test_legacy_format_owner_name(self, client, mock_model_response):
+        """Test legacy format: models.get('owner/name')."""
+        # Mock the original get method
+        client.models._original_get = Mock(return_value=mock_model_response)
+
+        # Call with legacy format
+        result = client.models.get("stability-ai/stable-diffusion")
+
+        # Verify original method was called with correct parameters
+        client.models._original_get.assert_called_once_with(
+            model_owner="stability-ai",
+            model_name="stable-diffusion",
+            extra_headers=None,
+            extra_query=None,
+            extra_body=None,
+            timeout=NOT_GIVEN,
+        )
+        assert result == mock_model_response
+
+    def test_new_format_keyword_args(self, client, mock_model_response):
+        """Test new format: models.get(model_owner='owner', model_name='name')."""
+        # Mock the original get method
+        client.models._original_get = Mock(return_value=mock_model_response)
+
+        # Call with new format
+        result = client.models.get(model_owner="stability-ai", model_name="stable-diffusion")
+
+        # Verify original method was called with correct parameters
+        client.models._original_get.assert_called_once_with(
+            model_owner="stability-ai",
+            model_name="stable-diffusion",
+            extra_headers=None,
+            extra_query=None,
+            extra_body=None,
+            timeout=NOT_GIVEN,
+        )
+        assert result == mock_model_response
+
+    def test_legacy_format_with_extra_params(self, client, mock_model_response):
+        """Test legacy format with extra parameters."""
+        # Mock the original get method
+        client.models._original_get = Mock(return_value=mock_model_response)
+
+        # Call with legacy format and extra parameters
+        result = client.models.get("stability-ai/stable-diffusion", extra_headers={"X-Custom": "test"}, timeout=30.0)
+
+        # Verify original method was called with correct parameters
+        client.models._original_get.assert_called_once_with(
+            model_owner="stability-ai",
+            model_name="stable-diffusion",
+            extra_headers={"X-Custom": "test"},
+            extra_query=None,
+            extra_body=None,
+            timeout=30.0,
+        )
+        assert result == mock_model_response
+
+    def test_error_mixed_formats(self, client):
+        """Test error when mixing legacy and new formats."""
+        with pytest.raises(ValueError) as exc_info:
+            client.models.get("stability-ai/stable-diffusion", model_owner="other-owner")
+
+        assert "Cannot specify both positional 'model_or_owner' and keyword arguments" in str(exc_info.value)
+
+    def test_error_invalid_legacy_format(self, client):
+        """Test error for invalid legacy format (no slash)."""
+        with pytest.raises(ValueError) as exc_info:
+            client.models.get("invalid-format")
+
+        assert "Invalid model reference 'invalid-format'" in str(exc_info.value)
+        assert "Expected format: 'owner/name'" in str(exc_info.value)
+
+    def test_error_missing_parameters(self, client):
+        """Test error when no parameters are provided."""
+        with pytest.raises(ValueError) as exc_info:
+            client.models.get()
+
+        assert "model_owner and model_name are required" in str(exc_info.value)
+
+    def test_legacy_format_with_complex_names(self, client, mock_model_response):
+        """Test legacy format with complex owner/model names."""
+        # Mock the original get method
+        client.models._original_get = Mock(return_value=mock_model_response)
+
+        # Test with hyphenated names and numbers
+        result = client.models.get("black-forest-labs/flux-1.1-pro")
+
+        # Verify parsing
+        client.models._original_get.assert_called_once_with(
+            model_owner="black-forest-labs",
+            model_name="flux-1.1-pro",
+            extra_headers=None,
+            extra_query=None,
+            extra_body=None,
+            timeout=NOT_GIVEN,
+        )
+
+    def test_legacy_format_multiple_slashes_error(self, client):
+        """Test error for legacy format with multiple slashes."""
+        with pytest.raises(ValueError) as exc_info:
+            client.models.get("owner/name/version")
+
+        assert "Invalid model reference" in str(exc_info.value)
+
+
+class TestAsyncModelGetBackwardCompatibility:
+    """Test backward compatibility for async models.get() method."""
+
+    @pytest.fixture
+    async def async_client(self):
+        """Create an async Replicate client with mocked token."""
+        with patch("replicate.lib.cog._get_api_token_from_environment", return_value="test-token"):
+            return AsyncReplicate()
+
+    @pytest.mark.asyncio
+    async def test_async_legacy_format_owner_name(self, async_client, mock_model_response):
+        """Test async legacy format: models.get('owner/name')."""
+        # Mock the original async get method
+        async_client.models._original_get = AsyncMock(return_value=mock_model_response)
+
+        # Call with legacy format
+        result = await async_client.models.get("stability-ai/stable-diffusion")
+
+        # Verify original method was called with correct parameters
+        async_client.models._original_get.assert_called_once_with(
+            model_owner="stability-ai",
+            model_name="stable-diffusion",
+            extra_headers=None,
+            extra_query=None,
+            extra_body=None,
+            timeout=NOT_GIVEN,
+        )
+        assert result == mock_model_response
+
+    @pytest.mark.asyncio
+    async def test_async_new_format_keyword_args(self, async_client, mock_model_response):
+        """Test async new format: models.get(model_owner='owner', model_name='name')."""
+        # Mock the original async get method
+        async_client.models._original_get = AsyncMock(return_value=mock_model_response)
+
+        # Call with new format
+        result = await async_client.models.get(model_owner="stability-ai", model_name="stable-diffusion")
+
+        # Verify original method was called with correct parameters
+        async_client.models._original_get.assert_called_once_with(
+            model_owner="stability-ai",
+            model_name="stable-diffusion",
+            extra_headers=None,
+            extra_query=None,
+            extra_body=None,
+            timeout=NOT_GIVEN,
+        )
+        assert result == mock_model_response
+
+    @pytest.mark.asyncio
+    async def test_async_error_mixed_formats(self, async_client):
+        """Test async error when mixing legacy and new formats."""
+        with pytest.raises(ValueError) as exc_info:
+            await async_client.models.get("stability-ai/stable-diffusion", model_owner="other-owner")
+
+        assert "Cannot specify both positional 'model_or_owner' and keyword arguments" in str(exc_info.value)
+
+
+class TestModelVersionIdentifierIntegration:
+    """Test integration with ModelVersionIdentifier parsing."""
+
+    @pytest.fixture
+    def client(self):
+        """Create a Replicate client with mocked token."""
+        with patch("replicate.lib.cog._get_api_token_from_environment", return_value="test-token"):
+            return Replicate()
+
+    def test_legacy_format_parsing_edge_cases(self, client, mock_model_response):
+        """Test edge cases in legacy format parsing."""
+        # Mock the original get method
+        client.models._original_get = Mock(return_value=mock_model_response)
+
+        # Test various valid formats
+        test_cases = [
+            ("owner/name", "owner", "name"),
+            ("owner-with-hyphens/name-with-hyphens", "owner-with-hyphens", "name-with-hyphens"),
+            ("owner123/name456", "owner123", "name456"),
+            ("owner/name.with.dots", "owner", "name.with.dots"),
+        ]
+
+        for model_ref, expected_owner, expected_name in test_cases:
+            client.models._original_get.reset_mock()
+            client.models.get(model_ref)
+
+            client.models._original_get.assert_called_once_with(
+                model_owner=expected_owner,
+                model_name=expected_name,
+                extra_headers=None,
+                extra_query=None,
+                extra_body=None,
+                timeout=NOT_GIVEN,
+            )