Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .fernignore
Original file line number Diff line number Diff line change
@@ -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
15 changes: 10 additions & 5 deletions src/agoraio/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
}
Expand All @@ -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"]
11 changes: 9 additions & 2 deletions src/agoraio/core/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -31,4 +38,4 @@ def __dir__():
return sorted(lazy_attrs)


__all__ = ["File", "with_content_type"]
__all__ = ["Area", "Pool", "create_pool", "File", "with_content_type"]
204 changes: 204 additions & 0 deletions src/agoraio/core/domain.py
Original file line number Diff line number Diff line change
@@ -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)
Loading