poeditor_client.pyโข15.1 kB
"""POEditor API client for managing translations and localization."""
import asyncio
from typing import Any, Dict, List, Optional, Union
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
import logging
logger = logging.getLogger(__name__)
class POEditorError(Exception):
"""Base exception for POEditor API errors."""
pass
class POEditorClient:
"""Client for interacting with the POEditor API."""
BASE_URL = "https://api.poeditor.com/v2"
def __init__(self, api_token: str, timeout: int = 30):
"""Initialize the POEditor client.
Args:
api_token: POEditor API token
timeout: Request timeout in seconds
"""
self.api_token = api_token
self.timeout = timeout
self._session: Optional[httpx.AsyncClient] = None
async def __aenter__(self):
"""Async context manager entry."""
self._session = httpx.AsyncClient(timeout=self.timeout)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit."""
if self._session:
await self._session.aclose()
@property
def session(self) -> httpx.AsyncClient:
"""Get or create HTTP session."""
if self._session is None:
self._session = httpx.AsyncClient(timeout=self.timeout)
return self._session
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def _make_request(
self,
endpoint: str,
data: Optional[Dict[str, Any]] = None,
files: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Make a request to the POEditor API.
Args:
endpoint: API endpoint (without base URL)
data: Form data to send
files: Files to upload
Returns:
API response data
Raises:
POEditorError: If the API returns an error
"""
url = f"{self.BASE_URL}/{endpoint}"
# Prepare form data
form_data = {"api_token": self.api_token}
if data:
form_data.update(data)
try:
if files:
response = await self.session.post(url, data=form_data, files=files)
else:
response = await self.session.post(url, data=form_data)
response.raise_for_status()
result = response.json()
if result.get("response", {}).get("status") != "success":
error_msg = result.get("response", {}).get("message", "Unknown error")
raise POEditorError(f"API error: {error_msg}")
return result.get("result", {})
except httpx.HTTPError as e:
logger.error(f"HTTP error occurred: {e}")
raise POEditorError(f"HTTP error: {e}")
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise POEditorError(f"Unexpected error: {e}")
# Project methods
async def list_projects(self) -> List[Dict[str, Any]]:
"""List all projects."""
result = await self._make_request("projects/list")
return result.get("projects", [])
async def view_project(self, project_id: Union[str, int]) -> Dict[str, Any]:
"""View project details.
Args:
project_id: Project ID
Returns:
Project details
"""
data = {"id": str(project_id)}
result = await self._make_request("projects/view", data)
return result.get("project", {})
async def create_project(
self,
name: str,
description: Optional[str] = None
) -> Dict[str, Any]:
"""Create a new project.
Args:
name: Project name
description: Optional project description
Returns:
Created project details
"""
data = {"name": name}
if description:
data["description"] = description
result = await self._make_request("projects/add", data)
return result.get("project", {})
async def update_project(
self,
project_id: Union[str, int],
name: Optional[str] = None,
description: Optional[str] = None
) -> Dict[str, Any]:
"""Update project details.
Args:
project_id: Project ID
name: New project name
description: New project description
Returns:
Updated project details
"""
data = {"id": str(project_id)}
if name:
data["name"] = name
if description:
data["description"] = description
result = await self._make_request("projects/update", data)
return result.get("project", {})
async def delete_project(self, project_id: Union[str, int]) -> bool:
"""Delete a project.
Args:
project_id: Project ID
Returns:
True if successful
"""
data = {"id": str(project_id)}
await self._make_request("projects/delete", data)
return True
# Language methods
async def list_languages(self, project_id: Union[str, int]) -> List[Dict[str, Any]]:
"""List languages in a project.
Args:
project_id: Project ID
Returns:
List of languages
"""
data = {"id": str(project_id)}
result = await self._make_request("languages/list", data)
return result.get("languages", [])
async def add_language(
self,
project_id: Union[str, int],
language_code: str
) -> Dict[str, Any]:
"""Add a language to a project.
Args:
project_id: Project ID
language_code: Language code (e.g., 'en', 'es', 'fr')
Returns:
Language details
"""
data = {
"id": str(project_id),
"language": language_code
}
result = await self._make_request("languages/add", data)
return result
async def delete_language(
self,
project_id: Union[str, int],
language_code: str
) -> bool:
"""Delete a language from a project.
Args:
project_id: Project ID
language_code: Language code
Returns:
True if successful
"""
data = {
"id": str(project_id),
"language": language_code
}
await self._make_request("languages/delete", data)
return True
# Terms methods
async def list_terms(
self,
project_id: Union[str, int],
language_code: Optional[str] = None
) -> List[Dict[str, Any]]:
"""List terms in a project.
Args:
project_id: Project ID
language_code: Optional language code to filter terms
Returns:
List of terms
"""
data = {"id": str(project_id)}
if language_code:
data["language"] = language_code
result = await self._make_request("terms/list", data)
return result.get("terms", [])
async def add_term(
self,
project_id: Union[str, int],
term: str,
context: Optional[str] = None,
reference: Optional[str] = None,
plural: Optional[str] = None
) -> Dict[str, Any]:
"""Add a term to a project.
Args:
project_id: Project ID
term: The term/key to add
context: Optional context for the term
reference: Optional reference for the term
plural: Optional plural form
Returns:
Term details
"""
data = {
"id": str(project_id),
"term": term
}
if context:
data["context"] = context
if reference:
data["reference"] = reference
if plural:
data["plural"] = plural
result = await self._make_request("terms/add", data)
return result
async def update_term(
self,
project_id: Union[str, int],
term_id: Union[str, int],
term: Optional[str] = None,
context: Optional[str] = None,
reference: Optional[str] = None,
plural: Optional[str] = None
) -> Dict[str, Any]:
"""Update a term.
Args:
project_id: Project ID
term_id: Term ID
term: New term value
context: New context
reference: New reference
plural: New plural form
Returns:
Updated term details
"""
data = {
"id": str(project_id),
"term_id": str(term_id)
}
if term:
data["term"] = term
if context:
data["context"] = context
if reference:
data["reference"] = reference
if plural:
data["plural"] = plural
result = await self._make_request("terms/update", data)
return result
async def delete_term(
self,
project_id: Union[str, int],
term_id: Union[str, int]
) -> bool:
"""Delete a term.
Args:
project_id: Project ID
term_id: Term ID
Returns:
True if successful
"""
data = {
"id": str(project_id),
"term_id": str(term_id)
}
await self._make_request("terms/delete", data)
return True
# Translation methods
async def list_translations(
self,
project_id: Union[str, int],
language_code: str
) -> List[Dict[str, Any]]:
"""List translations for a language.
Args:
project_id: Project ID
language_code: Language code
Returns:
List of translations
"""
data = {
"id": str(project_id),
"language": language_code
}
result = await self._make_request("translations/list", data)
return result.get("translations", [])
async def add_translation(
self,
project_id: Union[str, int],
language_code: str,
term_id: Union[str, int],
content: str,
fuzzy: bool = False
) -> Dict[str, Any]:
"""Add or update a translation.
Args:
project_id: Project ID
language_code: Language code
term_id: Term ID
content: Translation content
fuzzy: Whether the translation is fuzzy
Returns:
Translation details
"""
data = {
"id": str(project_id),
"language": language_code,
"term_id": str(term_id),
"content": content
}
if fuzzy:
data["fuzzy"] = "1"
result = await self._make_request("translations/add", data)
return result
async def update_translation(
self,
project_id: Union[str, int],
language_code: str,
term_id: Union[str, int],
content: str,
fuzzy: bool = False
) -> Dict[str, Any]:
"""Update a translation.
Args:
project_id: Project ID
language_code: Language code
term_id: Term ID
content: New translation content
fuzzy: Whether the translation is fuzzy
Returns:
Updated translation details
"""
data = {
"id": str(project_id),
"language": language_code,
"term_id": str(term_id),
"content": content
}
if fuzzy:
data["fuzzy"] = "1"
result = await self._make_request("translations/update", data)
return result
async def delete_translation(
self,
project_id: Union[str, int],
language_code: str,
term_id: Union[str, int]
) -> bool:
"""Delete a translation.
Args:
project_id: Project ID
language_code: Language code
term_id: Term ID
Returns:
True if successful
"""
data = {
"id": str(project_id),
"language": language_code,
"term_id": str(term_id)
}
await self._make_request("translations/delete", data)
return True
# Export methods
async def export_translations(
self,
project_id: Union[str, int],
language_code: str,
file_format: str = "json",
filters: Optional[List[str]] = None,
order: Optional[str] = None
) -> str:
"""Export translations.
Args:
project_id: Project ID
language_code: Language code
file_format: Export format (json, csv, xml, etc.)
filters: Optional filters to apply
order: Optional ordering
Returns:
URL to download the exported file
"""
data = {
"id": str(project_id),
"language": language_code,
"type": file_format
}
if filters:
data["filters"] = filters
if order:
data["order"] = order
result = await self._make_request("projects/export", data)
return result.get("url", "")
# Statistics methods
async def get_project_stats(self, project_id: Union[str, int]) -> Dict[str, Any]:
"""Get project statistics.
Args:
project_id: Project ID
Returns:
Project statistics
"""
data = {"id": str(project_id)}
result = await self._make_request("projects/view", data)
return result.get("project", {})
async def get_language_stats(
self,
project_id: Union[str, int],
language_code: str
) -> Dict[str, Any]:
"""Get language statistics.
Args:
project_id: Project ID
language_code: Language code
Returns:
Language statistics
"""
languages = await self.list_languages(project_id)
for lang in languages:
if lang.get("code") == language_code:
return lang
return {}
async def close(self):
"""Close the HTTP session."""
if self._session:
await self._session.aclose()