feat: add vNext engine support

src/lingodotdev/engine.py

@@ -18,15 +18,30 @@ 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("engine_id", pre=True, always=True)
  @classmethod
- def validate_api_url(cls, v: str) -> str:
+ def validate_engine_id(cls, v: Optional[str]) -> Optional[str]:
+ if v is None:
+ return None
+ v = v.strip()
+ return v if v else None
+
+ @validator("api_url", pre=True, always=True)
+ @classmethod
+ def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+ default_url = "https://engine.lingo.dev"
+ if v is None:
+ v = default_url
  if not v.startswith(("http://", "https://")):
  raise ValueError("API URL must be a valid HTTP/HTTPS URL")
+ v = v.rstrip("/")
+ if v == default_url and values.get("engine_id"):
+ return "https://api.lingo.dev"
  return v
 
 
@@ -55,6 +70,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 +88,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 +224,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 +490,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 +526,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 +587,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.
@@ -553,6 +600,7 @@ async def quick_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  Translated content (same type as input)
@@ -576,6 +624,8 @@ async def quick_translate(
  "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 +650,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.
@@ -612,6 +663,7 @@ async def quick_batch_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  List of translated content (one for each target locale)
@@ -628,6 +680,8 @@ async def quick_batch_translate(
  "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,180 @@ 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 teardown_method(self):
+ """Clean up engine client"""
+ if self.engine._client and not self.engine._client.is_closed:
+ asyncio.get_event_loop().run_until_complete(self.engine.close())
+
+ def test_engine_id_empty_string_treated_as_none(self):
+ """Test that empty engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": ""})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_whitespace_treated_as_none(self):
+ """Test that whitespace-only engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " "})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_stripped(self):
+ """Test that engine_id is stripped of whitespace"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " eng_123 "})
+ assert engine.config.engine_id == "eng_123"
+ assert engine._is_vnext is True
+
+ def test_api_url_trailing_slash_stripped(self):
+ """Test that trailing slash is stripped from api_url"""
+ 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_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

src/lingodotdev/engine.py

@@ -18,15 +18,30 @@ 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("engine_id", pre=True, always=True)
  @classmethod
- def validate_api_url(cls, v: str) -> str:
+ def validate_engine_id(cls, v: Optional[str]) -> Optional[str]:
+ if v is None:
+ return None
+ v = v.strip()
+ return v if v else None
+
+ @validator("api_url", pre=True, always=True)
+ @classmethod
+ def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+ default_url = "https://engine.lingo.dev"
+ if v is None:
+ v = default_url
  if not v.startswith(("http://", "https://")):
  raise ValueError("API URL must be a valid HTTP/HTTPS URL")
+ v = v.rstrip("/")
+ if v == default_url and values.get("engine_id"):
+ return "https://api.lingo.dev"
  return v
 
 
@@ -55,6 +70,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 +88,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 +224,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 +490,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 +526,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 +587,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.
@@ -553,6 +600,7 @@ async def quick_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  Translated content (same type as input)
@@ -576,6 +624,8 @@ async def quick_translate(
  "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 +650,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.
@@ -612,6 +663,7 @@ async def quick_batch_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  List of translated content (one for each target locale)
@@ -628,6 +680,8 @@ async def quick_batch_translate(
  "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,180 @@ 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 teardown_method(self):
+ """Clean up engine client"""
+ if self.engine._client and not self.engine._client.is_closed:
+ asyncio.get_event_loop().run_until_complete(self.engine.close())
+
+ def test_engine_id_empty_string_treated_as_none(self):
+ """Test that empty engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": ""})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_whitespace_treated_as_none(self):
+ """Test that whitespace-only engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " "})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_stripped(self):
+ """Test that engine_id is stripped of whitespace"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " eng_123 "})
+ assert engine.config.engine_id == "eng_123"
+ assert engine._is_vnext is True
+
+ def test_api_url_trailing_slash_stripped(self):
+ """Test that trailing slash is stripped from api_url"""
+ 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_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

src/lingodotdev/engine.py

@@ -18,15 +18,30 @@ 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("engine_id", pre=True, always=True)
  @classmethod
- def validate_api_url(cls, v: str) -> str:
+ def validate_engine_id(cls, v: Optional[str]) -> Optional[str]:
+ if v is None:
+ return None
+ v = v.strip()
+ return v if v else None
+
+ @validator("api_url", pre=True, always=True)
+ @classmethod
+ def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+ default_url = "https://engine.lingo.dev"
+ if v is None:
+ v = default_url
  if not v.startswith(("http://", "https://")):
  raise ValueError("API URL must be a valid HTTP/HTTPS URL")
+ v = v.rstrip("/")
+ if v == default_url and values.get("engine_id"):
+ return "https://api.lingo.dev"
  return v
 
 
@@ -55,6 +70,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 +88,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 +224,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 +490,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 +526,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 +587,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.
@@ -553,6 +600,7 @@ async def quick_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  Translated content (same type as input)
@@ -576,6 +624,8 @@ async def quick_translate(
  "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 +650,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.
@@ -612,6 +663,7 @@ async def quick_batch_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  List of translated content (one for each target locale)
@@ -628,6 +680,8 @@ async def quick_batch_translate(
  "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,180 @@ 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 teardown_method(self):
+ """Clean up engine client"""
+ if self.engine._client and not self.engine._client.is_closed:
+ asyncio.get_event_loop().run_until_complete(self.engine.close())
+
+ def test_engine_id_empty_string_treated_as_none(self):
+ """Test that empty engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": ""})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_whitespace_treated_as_none(self):
+ """Test that whitespace-only engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " "})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_stripped(self):
+ """Test that engine_id is stripped of whitespace"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " eng_123 "})
+ assert engine.config.engine_id == "eng_123"
+ assert engine._is_vnext is True
+
+ def test_api_url_trailing_slash_stripped(self):
+ """Test that trailing slash is stripped from api_url"""
+ 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_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

src/lingodotdev/engine.py

@@ -18,15 +18,30 @@ 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("engine_id", pre=True, always=True)
  @classmethod
- def validate_api_url(cls, v: str) -> str:
+ def validate_engine_id(cls, v: Optional[str]) -> Optional[str]:
+ if v is None:
+ return None
+ v = v.strip()
+ return v if v else None
+
+ @validator("api_url", pre=True, always=True)
+ @classmethod
+ def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+ default_url = "https://engine.lingo.dev"
+ if v is None:
+ v = default_url
  if not v.startswith(("http://", "https://")):
  raise ValueError("API URL must be a valid HTTP/HTTPS URL")
+ v = v.rstrip("/")
+ if v == default_url and values.get("engine_id"):
+ return "https://api.lingo.dev"
  return v
 
 
@@ -55,6 +70,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 +88,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 +224,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 +490,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 +526,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 +587,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.
@@ -553,6 +600,7 @@ async def quick_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  Translated content (same type as input)
@@ -576,6 +624,8 @@ async def quick_translate(
  "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 +650,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.
@@ -612,6 +663,7 @@ async def quick_batch_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  List of translated content (one for each target locale)
@@ -628,6 +680,8 @@ async def quick_batch_translate(
  "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,180 @@ 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 teardown_method(self):
+ """Clean up engine client"""
+ if self.engine._client and not self.engine._client.is_closed:
+ asyncio.get_event_loop().run_until_complete(self.engine.close())
+
+ def test_engine_id_empty_string_treated_as_none(self):
+ """Test that empty engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": ""})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_whitespace_treated_as_none(self):
+ """Test that whitespace-only engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " "})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_stripped(self):
+ """Test that engine_id is stripped of whitespace"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " eng_123 "})
+ assert engine.config.engine_id == "eng_123"
+ assert engine._is_vnext is True
+
+ def test_api_url_trailing_slash_stripped(self):
+ """Test that trailing slash is stripped from api_url"""
+ 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_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

src/lingodotdev/engine.py

@@ -18,15 +18,30 @@ 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("engine_id", pre=True, always=True)
  @classmethod
- def validate_api_url(cls, v: str) -> str:
+ def validate_engine_id(cls, v: Optional[str]) -> Optional[str]:
+ if v is None:
+ return None
+ v = v.strip()
+ return v if v else None
+
+ @validator("api_url", pre=True, always=True)
+ @classmethod
+ def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+ default_url = "https://engine.lingo.dev"
+ if v is None:
+ v = default_url
  if not v.startswith(("http://", "https://")):
  raise ValueError("API URL must be a valid HTTP/HTTPS URL")
+ v = v.rstrip("/")
+ if v == default_url and values.get("engine_id"):
+ return "https://api.lingo.dev"
  return v
 
 
@@ -55,6 +70,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 +88,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 +224,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 +490,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 +526,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 +587,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.
@@ -553,6 +600,7 @@ async def quick_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  Translated content (same type as input)
@@ -576,6 +624,8 @@ async def quick_translate(
  "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 +650,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.
@@ -612,6 +663,7 @@ async def quick_batch_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  List of translated content (one for each target locale)
@@ -628,6 +680,8 @@ async def quick_batch_translate(
  "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,180 @@ 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 teardown_method(self):
+ """Clean up engine client"""
+ if self.engine._client and not self.engine._client.is_closed:
+ asyncio.get_event_loop().run_until_complete(self.engine.close())
+
+ def test_engine_id_empty_string_treated_as_none(self):
+ """Test that empty engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": ""})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_whitespace_treated_as_none(self):
+ """Test that whitespace-only engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " "})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_stripped(self):
+ """Test that engine_id is stripped of whitespace"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " eng_123 "})
+ assert engine.config.engine_id == "eng_123"
+ assert engine._is_vnext is True
+
+ def test_api_url_trailing_slash_stripped(self):
+ """Test that trailing slash is stripped from api_url"""
+ 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_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

src/lingodotdev/engine.py

@@ -18,15 +18,30 @@ 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("engine_id", pre=True, always=True)
  @classmethod
- def validate_api_url(cls, v: str) -> str:
+ def validate_engine_id(cls, v: Optional[str]) -> Optional[str]:
+ if v is None:
+ return None
+ v = v.strip()
+ return v if v else None
+
+ @validator("api_url", pre=True, always=True)
+ @classmethod
+ def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+ default_url = "https://engine.lingo.dev"
+ if v is None:
+ v = default_url
  if not v.startswith(("http://", "https://")):
  raise ValueError("API URL must be a valid HTTP/HTTPS URL")
+ v = v.rstrip("/")
+ if v == default_url and values.get("engine_id"):
+ return "https://api.lingo.dev"
  return v
 
 
@@ -55,6 +70,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 +88,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 +224,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 +490,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 +526,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 +587,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.
@@ -553,6 +600,7 @@ async def quick_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  Translated content (same type as input)
@@ -576,6 +624,8 @@ async def quick_translate(
  "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 +650,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.
@@ -612,6 +663,7 @@ async def quick_batch_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  List of translated content (one for each target locale)
@@ -628,6 +680,8 @@ async def quick_batch_translate(
  "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,180 @@ 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 teardown_method(self):
+ """Clean up engine client"""
+ if self.engine._client and not self.engine._client.is_closed:
+ asyncio.get_event_loop().run_until_complete(self.engine.close())
+
+ def test_engine_id_empty_string_treated_as_none(self):
+ """Test that empty engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": ""})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_whitespace_treated_as_none(self):
+ """Test that whitespace-only engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " "})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_stripped(self):
+ """Test that engine_id is stripped of whitespace"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " eng_123 "})
+ assert engine.config.engine_id == "eng_123"
+ assert engine._is_vnext is True
+
+ def test_api_url_trailing_slash_stripped(self):
+ """Test that trailing slash is stripped from api_url"""
+ 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_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

src/lingodotdev/engine.py

@@ -18,15 +18,30 @@ 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("engine_id", pre=True, always=True)
  @classmethod
- def validate_api_url(cls, v: str) -> str:
+ def validate_engine_id(cls, v: Optional[str]) -> Optional[str]:
+ if v is None:
+ return None
+ v = v.strip()
+ return v if v else None
+
+ @validator("api_url", pre=True, always=True)
+ @classmethod
+ def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+ default_url = "https://engine.lingo.dev"
+ if v is None:
+ v = default_url
  if not v.startswith(("http://", "https://")):
  raise ValueError("API URL must be a valid HTTP/HTTPS URL")
+ v = v.rstrip("/")
+ if v == default_url and values.get("engine_id"):
+ return "https://api.lingo.dev"
  return v
 
 
@@ -55,6 +70,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 +88,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 +224,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 +490,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 +526,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 +587,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.
@@ -553,6 +600,7 @@ async def quick_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  Translated content (same type as input)
@@ -576,6 +624,8 @@ async def quick_translate(
  "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 +650,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.
@@ -612,6 +663,7 @@ async def quick_batch_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  List of translated content (one for each target locale)
@@ -628,6 +680,8 @@ async def quick_batch_translate(
  "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,180 @@ 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 teardown_method(self):
+ """Clean up engine client"""
+ if self.engine._client and not self.engine._client.is_closed:
+ asyncio.get_event_loop().run_until_complete(self.engine.close())
+
+ def test_engine_id_empty_string_treated_as_none(self):
+ """Test that empty engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": ""})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_whitespace_treated_as_none(self):
+ """Test that whitespace-only engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " "})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_stripped(self):
+ """Test that engine_id is stripped of whitespace"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " eng_123 "})
+ assert engine.config.engine_id == "eng_123"
+ assert engine._is_vnext is True
+
+ def test_api_url_trailing_slash_stripped(self):
+ """Test that trailing slash is stripped from api_url"""
+ 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_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

src/lingodotdev/engine.py

@@ -18,15 +18,30 @@ 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("engine_id", pre=True, always=True)
  @classmethod
- def validate_api_url(cls, v: str) -> str:
+ def validate_engine_id(cls, v: Optional[str]) -> Optional[str]:
+ if v is None:
+ return None
+ v = v.strip()
+ return v if v else None
+
+ @validator("api_url", pre=True, always=True)
+ @classmethod
+ def validate_api_url(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+ default_url = "https://engine.lingo.dev"
+ if v is None:
+ v = default_url
  if not v.startswith(("http://", "https://")):
  raise ValueError("API URL must be a valid HTTP/HTTPS URL")
+ v = v.rstrip("/")
+ if v == default_url and values.get("engine_id"):
+ return "https://api.lingo.dev"
  return v
 
 
@@ -55,6 +70,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 +88,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 +224,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 +490,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 +526,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 +587,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.
@@ -553,6 +600,7 @@ async def quick_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  Translated content (same type as input)
@@ -576,6 +624,8 @@ async def quick_translate(
  "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 +650,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.
@@ -612,6 +663,7 @@ async def quick_batch_translate(
  source_locale: Source language code (optional, auto-detected if None)
  api_url: API endpoint URL
  fast: Enable fast mode for quicker translations
+ engine_id: Optional engine ID for vNext API.
 
  Returns:
  List of translated content (one for each target locale)
@@ -628,6 +680,8 @@ async def quick_batch_translate(
  "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,180 @@ 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 teardown_method(self):
+ """Clean up engine client"""
+ if self.engine._client and not self.engine._client.is_closed:
+ asyncio.get_event_loop().run_until_complete(self.engine.close())
+
+ def test_engine_id_empty_string_treated_as_none(self):
+ """Test that empty engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": ""})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_whitespace_treated_as_none(self):
+ """Test that whitespace-only engine_id is treated as None"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " "})
+ assert engine.config.engine_id is None
+ assert engine._is_vnext is False
+ assert engine.config.api_url == "https://engine.lingo.dev"
+
+ def test_engine_id_stripped(self):
+ """Test that engine_id is stripped of whitespace"""
+ engine = LingoDotDevEngine({"api_key": "key", "engine_id": " eng_123 "})
+ assert engine.config.engine_id == "eng_123"
+ assert engine._is_vnext is True
+
+ def test_api_url_trailing_slash_stripped(self):
+ """Test that trailing slash is stripped from api_url"""
+ 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_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