Merge pull request #2 from fern-demo/devin/1767822715-area-based-pool

Add Area-based Pool logic for regional URL cycling

Author: fern-support

.fernignore

@@ -1 +1,4 @@
 # Specify files that shouldn't be modified by Fern
+src/agoraio/pool_client.py
+src/agoraio/__init__.py
+src/agoraio/core/domain.py

src/agoraio/__init__.py

@@ -6,14 +6,19 @@
 from importlib import import_module
 
 if typing.TYPE_CHECKING:
-    from . import agents, phone_numbers, telephony
-    from .client import Agora, AsyncAgora
+    from . import agents, core, phone_numbers, telephony
+    from .core.domain import Area, Pool, create_pool
+    from .pool_client import Agora, AsyncAgora
     from .version import __version__
 _dynamic_imports: typing.Dict[str, str] = {
-    "Agora": ".client",
-    "AsyncAgora": ".client",
+    "Agora": ".pool_client",
+    "Area": ".core.domain",
+    "AsyncAgora": ".pool_client",
+    "Pool": ".core.domain",
     "__version__": ".version",
     "agents": ".agents",
+    "core": ".core",
+    "create_pool": ".core.domain",
     "phone_numbers": ".phone_numbers",
     "telephony": ".telephony",
 }
@@ -40,4 +45,4 @@ def __dir__():
     return sorted(lazy_attrs)
 
 
-__all__ = ["Agora", "AsyncAgora", "__version__", "agents", "phone_numbers", "telephony"]
+__all__ = ["Agora", "Area", "AsyncAgora", "Pool", "__version__", "agents", "core", "create_pool", "phone_numbers", "telephony"]

src/agoraio/core/__init__.py

@@ -6,8 +6,15 @@
 from importlib import import_module
 
 if typing.TYPE_CHECKING:
+    from .domain import Area, Pool, create_pool
     from .file import File, with_content_type
-_dynamic_imports: typing.Dict[str, str] = {"File": ".file", "with_content_type": ".file"}
+_dynamic_imports: typing.Dict[str, str] = {
+    "Area": ".domain",
+    "Pool": ".domain",
+    "create_pool": ".domain",
+    "File": ".file",
+    "with_content_type": ".file",
+}
 
 
 def __getattr__(attr_name: str) -> typing.Any:
@@ -31,4 +38,4 @@ def __dir__():
     return sorted(lazy_attrs)
 
 
-__all__ = ["File", "with_content_type"]
+__all__ = ["Area", "Pool", "create_pool", "File", "with_content_type"]

src/agoraio/core/domain.py

@@ -0,0 +1,204 @@
+# This file was auto-generated by Fern from our API Definition.
+
+import asyncio
+import socket
+import threading
+import time
+from enum import IntEnum
+from typing import List, Optional
+
+
+class Area(IntEnum):
+    """Area represents the global regions where the Open API gateway endpoint is located"""
+
+    UNKNOWN = 0
+    US = 1  # US represents the western and eastern regions of the United States
+    EU = 2  # EU represents the western and central regions of Europe
+    AP = 3  # AP represents the southeastern and northeastern regions of Asia-Pacific
+    CN = 4  # CN represents the eastern and northern regions of Chinese mainland
+
+
+CHINESE_MAINLAND_MAJOR_DOMAIN = "sd-rtn.com"
+OVERSEA_MAJOR_DOMAIN = "agora.io"
+
+GLOBAL_DOMAIN_PREFIX = "api"
+
+US_WEST_REGION_DOMAIN_PREFIX = "api-us-west-1"
+US_EAST_REGION_DOMAIN_PREFIX = "api-us-east-1"
+
+AP_SOUTHEAST_REGION_DOMAIN_PREFIX = "api-ap-southeast-1"
+AP_NORTHEAST_REGION_DOMAIN_PREFIX = "api-ap-northeast-1"
+
+EU_WEST_REGION_DOMAIN_PREFIX = "api-eu-west-1"
+EU_CENTRAL_REGION_DOMAIN_PREFIX = "api-eu-central-1"
+
+CN_EAST_REGION_DOMAIN_PREFIX = "api-cn-east-1"
+CN_NORTH_REGION_DOMAIN_PREFIX = "api-cn-north-1"
+
+API_PATH_SUFFIX = "/api/conversational-ai-agent"
+
+
+class Domain:
+    """Domain contains the regional prefixes and domain suffixes for an area"""
+
+    def __init__(self, region_domain_prefixes: List[str], major_domain_suffixes: List[str]):
+        self.region_domain_prefixes = region_domain_prefixes
+        self.major_domain_suffixes = major_domain_suffixes
+
+
+REGION_DOMAIN = {
+    Area.UNKNOWN: Domain([], []),
+    Area.US: Domain(
+        [US_WEST_REGION_DOMAIN_PREFIX, US_EAST_REGION_DOMAIN_PREFIX],
+        [OVERSEA_MAJOR_DOMAIN, CHINESE_MAINLAND_MAJOR_DOMAIN],
+    ),
+    Area.EU: Domain(
+        [EU_WEST_REGION_DOMAIN_PREFIX, EU_CENTRAL_REGION_DOMAIN_PREFIX],
+        [OVERSEA_MAJOR_DOMAIN, CHINESE_MAINLAND_MAJOR_DOMAIN],
+    ),
+    Area.AP: Domain(
+        [AP_SOUTHEAST_REGION_DOMAIN_PREFIX, AP_NORTHEAST_REGION_DOMAIN_PREFIX],
+        [OVERSEA_MAJOR_DOMAIN, CHINESE_MAINLAND_MAJOR_DOMAIN],
+    ),
+    Area.CN: Domain(
+        [CN_EAST_REGION_DOMAIN_PREFIX, CN_NORTH_REGION_DOMAIN_PREFIX],
+        [CHINESE_MAINLAND_MAJOR_DOMAIN, OVERSEA_MAJOR_DOMAIN],
+    ),
+}
+
+
+class Resolver:
+    """Interface for resolving the best domain"""
+
+    def resolve(self, domains: List[str], region_prefix: str) -> str:
+        raise NotImplementedError
+
+
+class ResolverImpl(Resolver):
+    """Default DNS-based resolver implementation"""
+
+    def resolve(self, domains: List[str], region_prefix: str) -> str:
+        result: Optional[str] = None
+        result_lock = threading.Lock()
+
+        def lookup_domain(domain: str) -> None:
+            nonlocal result
+            try:
+                url = f"{region_prefix}.{domain}"
+                socket.gethostbyname(url)
+                with result_lock:
+                    if result is None:
+                        result = domain
+            except socket.gaierror:
+                pass
+
+        threads = []
+        for domain in domains:
+            thread = threading.Thread(target=lookup_domain, args=(domain,))
+            thread.start()
+            threads.append(thread)
+
+        for thread in threads:
+            thread.join(timeout=5.0)
+
+        if result is not None:
+            return result
+
+        raise Exception("query all dns failed")
+
+
+class AsyncResolverImpl(Resolver):
+    """Async DNS-based resolver implementation"""
+
+    async def resolve_async(self, domains: List[str], region_prefix: str) -> str:
+        async def lookup_domain(domain: str) -> str:
+            url = f"{region_prefix}.{domain}"
+            loop = asyncio.get_event_loop()
+            await loop.getaddrinfo(url, None)
+            return domain
+
+        tasks = [lookup_domain(domain) for domain in domains]
+
+        for coro in asyncio.as_completed(tasks):
+            try:
+                result = await coro
+                return result
+            except (socket.gaierror, OSError):
+                continue
+
+        raise Exception("query all dns failed")
+
+
+UPDATE_DURATION_SECONDS = 30
+
+
+class Pool:
+    """Pool manages a pool of regional URLs with automatic cycling and domain selection"""
+
+    def __init__(self, domain_area: Area):
+        domain_config = REGION_DOMAIN.get(domain_area)
+        if domain_config is None or len(domain_config.region_domain_prefixes) == 0:
+            raise ValueError("invalid domain area")
+
+        self._domain_area = domain_area
+        self._domain_suffixes = list(domain_config.major_domain_suffixes)
+        self._region_prefixes = list(domain_config.region_domain_prefixes)
+        self._current_region_prefixes = list(self._region_prefixes)
+        self._current_domain = self._domain_suffixes[0]
+        self._resolver = ResolverImpl()
+        self._async_resolver = AsyncResolverImpl()
+        self._last_update: float = 0
+        self._lock = threading.Lock()
+
+    def _domain_need_update(self) -> bool:
+        return time.time() - self._last_update > UPDATE_DURATION_SECONDS
+
+    def select_best_domain(self) -> None:
+        """SelectBestDomain uses DNS resolution to select the best available domain (sync)"""
+        if not self._domain_need_update():
+            return
+
+        with self._lock:
+            if self._domain_need_update():
+                domain = self._resolver.resolve(self._domain_suffixes, self._current_region_prefixes[0])
+                self._select_domain(domain)
+
+    async def select_best_domain_async(self) -> None:
+        """SelectBestDomain uses DNS resolution to select the best available domain (async)"""
+        if not self._domain_need_update():
+            return
+
+        with self._lock:
+            if self._domain_need_update():
+                domain = await self._async_resolver.resolve_async(
+                    self._domain_suffixes, self._current_region_prefixes[0]
+                )
+                self._select_domain(domain)
+
+    def next_region(self) -> None:
+        """NextRegion cycles to the next region prefix in the pool"""
+        with self._lock:
+            self._current_region_prefixes = self._current_region_prefixes[1:]
+            if len(self._current_region_prefixes) == 0:
+                self._current_region_prefixes = list(self._region_prefixes)
+
+    def _select_domain(self, domain: str) -> None:
+        if domain in self._domain_suffixes:
+            self._current_domain = domain
+            self._last_update = time.time()
+
+    def get_current_url(self) -> str:
+        """GetCurrentURL returns the current URL based on the selected region and domain"""
+        with self._lock:
+            current_region = self._current_region_prefixes[0]
+            current_domain = self._current_domain
+            return f"https://{current_region}.{current_domain}{API_PATH_SUFFIX}"
+
+    def get_area(self) -> Area:
+        """Get the current area"""
+        return self._domain_area
+
+
+def create_pool(area: Area) -> Pool:
+    """Creates a new Pool for the specified area"""
+    return Pool(area)