feat: add vNext engine support

Author: AndreyHirsa

src/lingodotdev/engine.py

@@ -18,13 +18,20 @@ class EngineConfig(BaseModel):
     """Configuration for the LingoDotDevEngine"""
 
     api_key: str
+    engine_id: Optional[str] = None
     api_url: str = "https://engine.lingo.dev"
     batch_size: int = Field(default=25, ge=1, le=250)
     ideal_batch_item_size: int = Field(default=250, ge=1, le=2500)
 
-    @validator("api_url")
+    @validator("api_url", pre=True, always=True)
     @classmethod
-    def validate_api_url(cls, v: str) -> str:
+    def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+        if v is None or v == "https://engine.lingo.dev":
+            engine_id = values.get("engine_id")
+            if engine_id:
+                return "https://api.lingo.dev"
+        if v is None:
+            return "https://engine.lingo.dev"
         if not v.startswith(("http://", "https://")):
             raise ValueError("API URL must be a valid HTTP/HTTPS URL")
         return v
@@ -55,6 +62,11 @@ def __init__(self, config: Dict[str, Any]):
         """
         self.config = EngineConfig(**config)
         self._client: Optional[httpx.AsyncClient] = None
+        self._session_id: str = generate()
+
+    @property
+    def _is_vnext(self) -> bool:
+        return self.config.engine_id is not None
 
     async def __aenter__(self):
         """Async context manager entry"""
@@ -68,10 +80,14 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
     async def _ensure_client(self):
         """Ensure the httpx client is initialized"""
         if self._client is None or self._client.is_closed:
+            if self._is_vnext:
+                auth_header = {"X-API-Key": self.config.api_key}
+            else:
+                auth_header = {"Authorization": f"Bearer {self.config.api_key}"}
             self._client = httpx.AsyncClient(
                 headers={
                     "Content-Type": "application/json; charset=utf-8",
-                    "Authorization": f"Bearer {self.config.api_key}",
+                    **auth_header,
                 },
                 timeout=60.0,
             )
@@ -200,16 +216,27 @@ async def _localize_chunk(
         """
         await self._ensure_client()
         assert self._client is not None  # Type guard for mypy
-        url = urljoin(self.config.api_url, "/i18n")
-
-        request_data = {
-            "params": {"workflowId": workflow_id, "fast": fast},
-            "locale": {"source": source_locale, "target": target_locale},
-            "data": payload["data"],
-        }
 
-        if payload.get("reference"):
-            request_data["reference"] = payload["reference"]
+        if self._is_vnext:
+            url = f"{self.config.api_url}/process/{self.config.engine_id}/localize"
+            request_data: Dict[str, Any] = {
+                "params": {"fast": fast},
+                "sourceLocale": source_locale,
+                "targetLocale": target_locale,
+                "data": payload["data"],
+                "sessionId": self._session_id,
+            }
+            if payload.get("reference"):
+                request_data["reference"] = payload["reference"]
+        else:
+            url = urljoin(self.config.api_url, "/i18n")
+            request_data = {
+                "params": {"workflowId": workflow_id, "fast": fast},
+                "locale": {"source": source_locale, "target": target_locale},
+                "data": payload["data"],
+            }
+            if payload.get("reference"):
+                request_data["reference"] = payload["reference"]
 
         try:
             response = await self._client.post(url, json=request_data)
@@ -455,7 +482,11 @@ async def recognize_locale(self, text: str) -> str:
 
         await self._ensure_client()
         assert self._client is not None  # Type guard for mypy
-        url = urljoin(self.config.api_url, "/recognize")
+
+        if self._is_vnext:
+            url = f"{self.config.api_url}/process/recognize"
+        else:
+            url = urljoin(self.config.api_url, "/recognize")
 
         try:
             response = await self._client.post(url, json={"text": text})
@@ -487,10 +518,17 @@ async def whoami(self) -> Optional[Dict[str, str]]:
         """
         await self._ensure_client()
         assert self._client is not None  # Type guard for mypy
-        url = urljoin(self.config.api_url, "/whoami")
+
+        if self._is_vnext:
+            url = f"{self.config.api_url}/users/me"
+        else:
+            url = urljoin(self.config.api_url, "/whoami")
 
         try:
-            response = await self._client.post(url)
+            if self._is_vnext:
+                response = await self._client.get(url)
+            else:
+                response = await self._client.post(url)
 
             if response.is_success:
                 payload = self._safe_parse_json(response)
@@ -541,6 +579,7 @@ async def quick_translate(
         source_locale: Optional[str] = None,
         api_url: str = "https://engine.lingo.dev",
         fast: bool = True,
+        engine_id: Optional[str] = None,
     ) -> Any:
         """
         Quick one-off translation without manual context management.
@@ -572,10 +611,12 @@ async def quick_translate(
                 "es"
             )
         """
-        config = {
+        config: Dict[str, Any] = {
             "api_key": api_key,
             "api_url": api_url,
         }
+        if engine_id:
+            config["engine_id"] = engine_id
 
         async with cls(config) as engine:
             params = {
@@ -600,6 +641,7 @@ async def quick_batch_translate(
         source_locale: Optional[str] = None,
         api_url: str = "https://engine.lingo.dev",
         fast: bool = True,
+        engine_id: Optional[str] = None,
     ) -> List[Any]:
         """
         Quick batch translation to multiple target locales.
@@ -624,10 +666,12 @@ async def quick_batch_translate(
             )
             # Results: ["Hola mundo", "Bonjour le monde", "Hallo Welt"]
         """
-        config = {
+        config: Dict[str, Any] = {
             "api_key": api_key,
             "api_url": api_url,
         }
+        if engine_id:
+            config["engine_id"] = engine_id
 
         async with cls(config) as engine:
             if isinstance(content, str):

tests/test_engine.py

@@ -641,3 +641,134 @@ async def test_full_localization_workflow(self, mock_post):
         assert request_data["locale"]["target"] == "es"
         assert request_data["params"]["fast"] is True
         assert request_data["data"] == {"greeting": "hello", "farewell": "goodbye"}
+
+
+@pytest.mark.asyncio
+class TestVNextEngine:
+    """Test vNext / Engine ID specific behavior"""
+
+    def setup_method(self):
+        """Set up test fixtures"""
+        self.config = {"api_key": "test_api_key", "engine_id": "my-engine-id"}
+        self.engine = LingoDotDevEngine(self.config)
+
+    def test_engine_id_default_api_url(self):
+        """Test that engine_id switches default api_url to api.lingo.dev"""
+        assert self.engine.config.api_url == "https://api.lingo.dev"
+        assert self.engine.config.engine_id == "my-engine-id"
+
+    def test_engine_id_with_explicit_api_url(self):
+        """Test that explicit api_url is preserved with engine_id"""
+        engine = LingoDotDevEngine({
+            "api_key": "key",
+            "engine_id": "eng",
+            "api_url": "https://custom.api.com",
+        })
+        assert engine.config.api_url == "https://custom.api.com"
+
+    def test_is_vnext_true(self):
+        """Test _is_vnext is True with engine_id"""
+        assert self.engine._is_vnext is True
+
+    def test_is_vnext_false_without_engine_id(self):
+        """Test _is_vnext is False without engine_id"""
+        engine = LingoDotDevEngine({"api_key": "key", "api_url": "https://api.test.com"})
+        assert engine._is_vnext is False
+
+    def test_session_id_generated(self):
+        """Test that session_id is generated on init"""
+        assert self.engine._session_id
+        assert isinstance(self.engine._session_id, str)
+
+    async def test_vnext_ensure_client_uses_x_api_key(self):
+        """Test that vNext engine uses X-API-Key header"""
+        await self.engine._ensure_client()
+        assert self.engine._client is not None
+        assert self.engine._client.headers.get("x-api-key") == "test_api_key"
+        assert "authorization" not in self.engine._client.headers
+        await self.engine.close()
+
+    async def test_classic_ensure_client_uses_bearer(self):
+        """Test that classic engine uses Bearer auth header"""
+        engine = LingoDotDevEngine({"api_key": "test_key", "api_url": "https://api.test.com"})
+        await engine._ensure_client()
+        assert engine._client is not None
+        assert engine._client.headers.get("authorization") == "Bearer test_key"
+        assert "x-api-key" not in engine._client.headers
+        await engine.close()
+
+    @patch("lingodotdev.engine.httpx.AsyncClient.post")
+    async def test_vnext_localize_chunk_url_and_body(self, mock_post):
+        """Test vNext localize chunk uses correct URL and body format"""
+        mock_response = Mock()
+        mock_response.is_success = True
+        mock_response.json.return_value = {"data": {"key": "translated"}}
+        mock_post.return_value = mock_response
+
+        await self.engine._localize_chunk(
+            "en", "es", {"data": {"key": "value"}, "reference": {"es": {"key": "ref"}}}, "wf", True
+        )
+
+        call_args = mock_post.call_args
+        url = call_args[0][0]
+        assert url == "https://api.lingo.dev/process/my-engine-id/localize"
+
+        body = call_args[1]["json"]
+        assert body["sourceLocale"] == "en"
+        assert body["targetLocale"] == "es"
+        assert body["params"] == {"fast": True}
+        assert body["data"] == {"key": "value"}
+        assert body["sessionId"] == self.engine._session_id
+        assert body["reference"] == {"es": {"key": "ref"}}
+
+    @patch("lingodotdev.engine.httpx.AsyncClient.post")
+    async def test_vnext_recognize_locale_url(self, mock_post):
+        """Test vNext recognize_locale uses correct URL"""
+        mock_response = Mock()
+        mock_response.is_success = True
+        mock_response.json.return_value = {"locale": "es"}
+        mock_post.return_value = mock_response
+
+        await self.engine.recognize_locale("Hola mundo")
+
+        url = mock_post.call_args[0][0]
+        assert url == "https://api.lingo.dev/process/recognize"
+
+    @patch("lingodotdev.engine.httpx.AsyncClient.get")
+    async def test_vnext_whoami(self, mock_get):
+        """Test vNext whoami calls GET /users/me"""
+        mock_response = Mock()
+        mock_response.is_success = True
+        mock_response.json.return_value = {"id": "usr_abc", "email": "user@example.com"}
+        mock_get.return_value = mock_response
+
+        result = await self.engine.whoami()
+
+        assert result == {"email": "user@example.com", "id": "usr_abc"}
+        url = mock_get.call_args[0][0]
+        assert url == "https://api.lingo.dev/users/me"
+
+    @patch("lingodotdev.engine.httpx.AsyncClient.post")
+    async def test_vnext_full_localization_workflow(self, mock_post):
+        """Test full vNext localization workflow via localize_object"""
+        mock_response = Mock()
+        mock_response.is_success = True
+        mock_response.json.return_value = {"data": {"greeting": "hola"}}
+        mock_post.return_value = mock_response
+
+        result = await self.engine.localize_object(
+            {"greeting": "hello"},
+            {"source_locale": "en", "target_locale": "es", "fast": True},
+        )
+
+        assert result == {"greeting": "hola"}
+
+        call_args = mock_post.call_args
+        url = call_args[0][0]
+        assert url == "https://api.lingo.dev/process/my-engine-id/localize"
+
+        body = call_args[1]["json"]
+        assert body["sourceLocale"] == "en"
+        assert body["targetLocale"] == "es"
+        assert "sessionId" in body
+        assert "locale" not in body  # classic format should NOT be present