client

client

Main CocApi client - simplified and modular version

CocApi

CocApi(
    token: str,
    timeout: int = 20,
    status_code: bool = False,
    config: ApiConfig | None = None,
    async_mode: bool = False,
)

Bases: ApiMethods

Clash of Clans API Wrapper with enhanced v3.0.0 features

Provides both sync and async interfaces with caching, metrics, middleware, and dynamic model generation capabilities.

Initialize CocApi with enhanced features

PARAMETER	DESCRIPTION
token	API token from developer.clashofclans.com TYPE: str
timeout	Request timeout in seconds (backward compatibility) TYPE: int DEFAULT: 20
status_code	Include status code in responses (backward compatibility) TYPE: bool DEFAULT: False
config	Optional ApiConfig for advanced settings TYPE: ApiConfig | None DEFAULT: None
async_mode	Enable async mode (default: False for backward compatibility) TYPE: bool DEFAULT: False

Source code in cocapi/client.py

def __init__(
    self,
    token: str,
    timeout: int = 20,
    status_code: bool = False,
    config: ApiConfig | None = None,
    async_mode: bool = False,
):
    """
    Initialize CocApi with enhanced features

    Args:
        token: API token from developer.clashofclans.com
        timeout: Request timeout in seconds (backward compatibility)
        status_code: Include status code in responses (backward compatibility)
        config: Optional ApiConfig for advanced settings
        async_mode: Enable async mode (default: False for backward compatibility)
    """
    self.token = token
    self.timeout = timeout
    self.status_code = status_code
    self.config = config or ApiConfig(timeout=timeout)
    self.async_mode = async_mode

    # Use config timeout if provided, otherwise use parameter
    if config is None:
        self.config.timeout = timeout

    self.ENDPOINT = self.config.base_url  # Backward compatibility
    self.headers = {
        "authorization": f"Bearer {token}",
        "Accept": "application/json",
    }

    # Initialize components
    self.cache = CacheManager(default_ttl=self.config.cache_ttl)
    self.cache.enable() if self.config.enable_caching else self.cache.disable()

    self.metrics = MetricsTracker(max_metrics=self.config.metrics_window_size)
    self.metrics.enable() if self.config.enable_metrics else self.metrics.disable()

    self.middleware = MiddlewareManager()

    # Async core for async operations
    self._async_core: AsyncCocApiCore | None = None

    # Key manager state (set by from_credentials())
    self._km_email: str | None = None
    self._km_password: str | None = None
    self._km_tokens: list[str] = []

    self.DEFAULT_PARAMS = ("limit", "after", "before")
    self.ERROR_INVALID_PARAM = {
        "result": "error",
        "message": "Invalid params for method",
    }

    # Only test sync mode immediately (async mode tests in context manager)
    if not async_mode:
        test_response = self.test()
        if test_response.get("result") == "error":
            raise ValueError(
                f"API initialization failed: {test_response.get('message')}"
            )

from_credentials classmethod

from_credentials(
    email: str,
    password: str,
    timeout: int = 20,
    status_code: bool = False,
    config: ApiConfig | None = None,
) -> CocApi

Create a CocApi instance using SuperCell developer portal credentials.

Instead of providing a raw API token, provide your developer portal email and password. Keys are automatically created and managed based on your current public IP.

PARAMETER	DESCRIPTION
email	SuperCell developer portal email TYPE: str
password	SuperCell developer portal password TYPE: str
timeout	Request timeout in seconds TYPE: int DEFAULT: 20
status_code	Include status code in responses TYPE: bool DEFAULT: False
config	Optional ApiConfig for advanced settings TYPE: ApiConfig | None DEFAULT: None

RETURNS	DESCRIPTION
CocApi	Configured CocApi instance with a managed API token

Example::

api = CocApi.from_credentials("email@example.com", "password")
clan = api.clan_tag("#CLAN_TAG")

Source code in cocapi/client.py

@classmethod
def from_credentials(
    cls,
    email: str,
    password: str,
    timeout: int = 20,
    status_code: bool = False,
    config: ApiConfig | None = None,
) -> "CocApi":
    """
    Create a CocApi instance using SuperCell developer portal credentials.

    Instead of providing a raw API token, provide your developer portal
    email and password. Keys are automatically created and managed
    based on your current public IP.

    Args:
        email: SuperCell developer portal email
        password: SuperCell developer portal password
        timeout: Request timeout in seconds
        status_code: Include status code in responses
        config: Optional ApiConfig for advanced settings

    Returns:
        Configured CocApi instance with a managed API token

    Example::

        api = CocApi.from_credentials("email@example.com", "password")
        clan = api.clan_tag("#CLAN_TAG")
    """
    from .key_manager import SyncKeyManager

    config = config or ApiConfig(timeout=timeout)

    with SyncKeyManager(
        email=email,
        password=password,
        key_name=config.key_name,
        key_count=config.key_count,
        key_description=config.key_description,
        persist_keys=config.persist_keys,
        key_storage_path=config.key_storage_path,
    ) as km:
        tokens = km.manage_keys()

    instance = cls(
        token=tokens[0],
        timeout=timeout,
        status_code=status_code,
        config=config,
    )
    instance._km_email = email
    instance._km_password = password
    instance._km_tokens = tokens
    return instance

__aenter__ async

__aenter__() -> CocApi

Async context manager entry - enables async mode automatically

Source code in cocapi/client.py

async def __aenter__(self) -> "CocApi":
    """Async context manager entry - enables async mode automatically"""
    if not self.async_mode:
        self.async_mode = True  # Auto-enable async mode in context

    self._async_core = AsyncCocApiCore(self.token, self.config)
    await self._async_core.__aenter__()

    # Propagate key manager state for auto-refresh
    if self._km_email is not None:
        self._async_core.set_key_manager_state(
            email=self._km_email,
            password=self._km_password,  # type: ignore[arg-type]
            key_name=self.config.key_name,
            key_count=self.config.key_count,
            auto_refresh=self.config.auto_refresh_keys,
            persist_keys=self.config.persist_keys,
            key_storage_path=self.config.key_storage_path,
        )

    # Test API connection in async mode
    test_response = await self._async_core.test_connection()
    if test_response.get("result") == "error":
        await self.__aexit__(None, None, None)
        raise ValueError(
            f"Async API initialization failed: {test_response.get('message')}"
        )

    return self

__aexit__ async

__aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Async context manager exit

Source code in cocapi/client.py

async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Async context manager exit"""
    if self._async_core:
        await self._async_core.__aexit__(exc_type, exc_val, exc_tb)
        self._async_core = None

test

test() -> dict[str, Any]

Test API connection

Source code in cocapi/client.py

def test(self) -> dict[str, Any]:
    """Test API connection"""
    try:
        response = self._sync_api_response("/locations")
        if response.get("result") == "error":
            return response
        return {"result": "success", "message": "API connection successful"}
    except Exception as e:
        return {
            "result": "error",
            "message": f"Connection test failed: {str(e)}",
            "error_type": "connection",
        }

paginate

paginate(
    method: Callable[..., Any], *args: Any, limit: int = 100
) -> Any

Auto-paginate through all results from a list endpoint.

Yields individual items from each page, automatically following the after cursor until all pages are exhausted.

PARAMETER	DESCRIPTION
method	An API method that returns paginated results (e.g. api.clan_members). TYPE: Callable[..., Any]
*args	Positional arguments for the method excluding the params dict (e.g. the clan tag). TYPE: Any DEFAULT: ()
limit	Items per page (default 100). TYPE: int DEFAULT: 100

RETURNS	DESCRIPTION
Any	Generator (sync) or async generator (async) of individual items.

Example::

for member in api.paginate(api.clan_members, "#TAG"):
    print(member["name"])

Source code in cocapi/client.py

def paginate(
    self,
    method: Callable[..., Any],
    *args: Any,
    limit: int = 100,
) -> Any:
    """Auto-paginate through all results from a list endpoint.

    Yields individual items from each page, automatically following
    the ``after`` cursor until all pages are exhausted.

    Args:
        method: An API method that returns paginated results
                (e.g. ``api.clan_members``).
        *args:  Positional arguments for the method **excluding** the
                ``params`` dict (e.g. the clan tag).
        limit:  Items per page (default 100).

    Returns:
        Generator (sync) or async generator (async) of individual items.

    Example::

        for member in api.paginate(api.clan_members, "#TAG"):
            print(member["name"])
    """
    if self.async_mode:
        return self._apaginate(method, *args, limit=limit)
    return self._paginate(method, *args, limit=limit)

batch

batch(
    method: Callable[..., Any],
    args_list: list[Any],
    max_concurrent: int | None = None,
) -> list[dict[str, Any]] | Awaitable[list[dict[str, Any]]]

Fetch multiple resources in one call.

PARAMETER	DESCRIPTION
method	An API method (e.g. api.players). TYPE: Callable[..., Any]
args_list	List of arguments — each element is passed to method. Use tuples for methods that take multiple positional args. TYPE: list[Any]
max_concurrent	Limit concurrent requests in async mode (ignored in sync mode). TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
list[dict[str, Any]] | Awaitable[list[dict[str, Any]]]	List of response dicts (sync) or awaitable list (async).
list[dict[str, Any]] | Awaitable[list[dict[str, Any]]]	Failed calls return their error dict in-place.

Example::

results = api.batch(api.players, ["#TAG1", "#TAG2", "#TAG3"])

Source code in cocapi/client.py

def batch(
    self,
    method: Callable[..., Any],
    args_list: list[Any],
    max_concurrent: int | None = None,
) -> list[dict[str, Any]] | Awaitable[list[dict[str, Any]]]:
    """Fetch multiple resources in one call.

    Args:
        method:         An API method (e.g. ``api.players``).
        args_list:      List of arguments — each element is passed to
                        *method*.  Use tuples for methods that take
                        multiple positional args.
        max_concurrent: Limit concurrent requests in async mode
                        (ignored in sync mode).

    Returns:
        List of response dicts (sync) or awaitable list (async).
        Failed calls return their error dict in-place.

    Example::

        results = api.batch(api.players, ["#TAG1", "#TAG2", "#TAG3"])
    """
    if self.async_mode:
        return self._abatch(method, args_list, max_concurrent)
    return self._batch(method, args_list)

custom_endpoint

custom_endpoint(
    endpoint_path: str,
    params: dict[str, Any] | None = None,
    use_dynamic_model: bool = False,
) -> dict[str, Any] | Awaitable[dict[str, Any]]

Call a custom API endpoint (future-proofing for new SuperCell endpoints)

PARAMETER	DESCRIPTION
endpoint_path	The endpoint path (e.g., "/clans/new-feature") TYPE: str
params	Optional query parameters TYPE: dict[str, Any] | None DEFAULT: None
use_dynamic_model	Whether to generate dynamic Pydantic models TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
dict[str, Any] | Awaitable[dict[str, Any]]	API response as dict or Pydantic model

Source code in cocapi/client.py

def custom_endpoint(
    self,
    endpoint_path: str,
    params: dict[str, Any] | None = None,
    use_dynamic_model: bool = False,
) -> dict[str, Any] | Awaitable[dict[str, Any]]:
    """
    Call a custom API endpoint (future-proofing for new SuperCell endpoints)

    Args:
        endpoint_path: The endpoint path (e.g., "/clans/new-feature")
        params: Optional query parameters
        use_dynamic_model: Whether to generate dynamic Pydantic models

    Returns:
        API response as dict or Pydantic model
    """
    if not endpoint_path.startswith("/"):
        endpoint_path = "/" + endpoint_path

    return self._api_response(endpoint_path, params, use_dynamic_model)

set_base_url

set_base_url(
    new_base_url: str, force: bool = False
) -> None

Change the base API URL with safety warnings

PARAMETER	DESCRIPTION
new_base_url	New base URL to use TYPE: str
force	Skip safety warnings (use with caution) TYPE: bool DEFAULT: False

Source code in cocapi/client.py

def set_base_url(self, new_base_url: str, force: bool = False) -> None:
    """
    Change the base API URL with safety warnings

    Args:
        new_base_url: New base URL to use
        force: Skip safety warnings (use with caution)
    """
    official_url = "https://api.clashofclans.com/v1"

    if new_base_url != official_url and not force:
        warn(
            f"⚠️  WARNING: Changing base URL from official endpoint!\n"
            f"   Official: {official_url}\n"
            f"   New URL:  {new_base_url}\n"
            f"   This may result in API failures or unexpected behavior.\n"
            f"   Use force=True to suppress this warning.",
            UserWarning,
            stacklevel=2,
        )

    original_url = self.config.base_url
    self.config.base_url = new_base_url
    self.ENDPOINT = new_base_url  # Update backward compatibility field

    logging.info(f"Base URL changed from {original_url} to {new_base_url}")

get_base_url

get_base_url() -> str

Get current base URL

Source code in cocapi/client.py

def get_base_url(self) -> str:
    """Get current base URL"""
    return self.config.base_url

reset_base_url

reset_base_url() -> None

Reset to official SuperCell API URL

Source code in cocapi/client.py

def reset_base_url(self) -> None:
    """Reset to official SuperCell API URL"""
    official_url = "https://api.clashofclans.com/v1"
    original_url = self.config.base_url

    self.config.base_url = official_url
    self.ENDPOINT = official_url

    if original_url != official_url:
        warn(
            f"Base URL reset to official SuperCell endpoint: {official_url}",
            UserWarning,
            stacklevel=2,
        )
        logging.info(f"Base URL reset from {original_url} to {official_url}")
    else:
        logging.info("Base URL is already set to the official endpoint")

add_request_middleware

add_request_middleware(
    middleware: Callable[
        [str, dict[str, str], dict[str, Any]],
        tuple[str, dict[str, str], dict[str, Any]],
    ],
) -> None

Add request middleware - delegates to middleware manager

Source code in cocapi/client.py

def add_request_middleware(
    self,
    middleware: Callable[
        [str, dict[str, str], dict[str, Any]],
        tuple[str, dict[str, str], dict[str, Any]],
    ],
) -> None:
    """Add request middleware - delegates to middleware manager"""
    self.middleware.add_request_middleware(middleware)

add_response_middleware

add_response_middleware(
    middleware: Callable[[dict[str, Any]], dict[str, Any]],
) -> None

Add response middleware - delegates to middleware manager

Source code in cocapi/client.py

def add_response_middleware(
    self, middleware: Callable[[dict[str, Any]], dict[str, Any]]
) -> None:
    """Add response middleware - delegates to middleware manager"""
    self.middleware.add_response_middleware(middleware)

get_metrics

get_metrics() -> dict[str, Any]

Get API metrics summary

Source code in cocapi/client.py

def get_metrics(self) -> dict[str, Any]:
    """Get API metrics summary"""
    return self.metrics.get_metrics_summary()

clear_metrics

clear_metrics() -> None

Clear stored metrics

Source code in cocapi/client.py

def clear_metrics(self) -> None:
    """Clear stored metrics"""
    self.metrics.clear_metrics()

clear_cache

clear_cache() -> int

Clear API response cache

Source code in cocapi/client.py

def clear_cache(self) -> int:
    """Clear API response cache"""
    return self.cache.clear()

get_cache_stats

get_cache_stats() -> dict[str, Any]

Get cache statistics

Source code in cocapi/client.py

def get_cache_stats(self) -> dict[str, Any]:
    """Get cache statistics"""
    return self.cache.get_stats()