Feat/q1 2026 2#38
Conversation
There was a problem hiding this comment.
Pull request overview
This PR upgrades cocapi to v4.0.0, modernizing the library around Python 3.10+ and adding major new capabilities: credential-based authentication with automatic key management, optional Pydantic response models, new/updated endpoints, and a pagination helper.
Changes:
- Add developer-portal credential auth via a new KeyManager (sync + async) with optional local key persistence and auto-refresh on invalid IP.
- Introduce optional Pydantic schemas + endpoint-to-model registry, plus new endpoints/wrappers (POST
verify_player_token, deprecated endpoint handling, pagination helper). - Bump packaging/tooling to Python 3.10+, update docs, and add endpoint tests + a live endpoint verification script.
Reviewed changes
Copilot reviewed 18 out of 20 changed files in this pull request and generated 12 comments.
Show a summary per file
| File | Description |
|---|---|
tests/test_cocapi.py |
Adds tests for deprecated endpoint wrapper behavior and new POST endpoint (verify_player_token). |
scripts/test_all_endpoints.py |
New live API endpoint verification script (excluded from build). |
pyproject.toml |
Version bump to 4.0.0, Python >=3.10, packaging exclusions, dependency adjustments. |
mypy.ini |
Updates mypy target Python version to 3.10. |
cocapi/utils.py |
Adds pagination helpers to extract items/cursors across dict/model responses. |
cocapi/schemas.py |
Adds concrete Pydantic response models (optional extra). |
cocapi/models.py |
Enhances model creation to prefer concrete schemas via a registry, fallback to dynamic models. |
cocapi/model_registry.py |
Introduces endpoint pattern → schema mapping and response resolution. |
cocapi/middleware.py |
Updates typing to Python 3.10 style generics/unions. |
cocapi/metrics.py |
Updates typing to Python 3.10 style generics/unions. |
cocapi/key_manager.py |
Implements sync/async developer portal key management, IP detection, and persistence. |
cocapi/config.py |
Adds KeyManager configuration knobs to ApiConfig. |
cocapi/client.py |
Adds credential constructor, auto-refresh on invalid IP, POST support, deprecated wrappers, pagination helper, and model auto-enable behavior. |
cocapi/cache.py |
Updates typing to Python 3.10 style generics/unions. |
cocapi/async_client.py |
Adds async token refresh support and async POST support. |
cocapi/api_methods.py |
Expands/updates endpoint methods (filters for clan search, POST verify token, new endpoints, deprecated wrappers). |
cocapi/__init__.py |
Exposes new KeyManager APIs and conditionally exports Pydantic models; bumps version constant. |
README.md |
Rewrites docs for v4 features (auth modes, pagination, configuration, Pydantic models). |
.gitignore |
Ignores .cocapi/ key storage directory. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| if result.get("result") == "error": | ||
| print(result["message"]) # Human-readable message | ||
| print(result["error_type"]) # "timeout", "connection", "http", "json", "retry_exhausted", "unknown" | ||
| print(result.get("status_code")) # HTTP status code if applicable |
There was a problem hiding this comment.
The error-handling example suggests status_code is present on error responses, but the implementation uses http_status for HTTP errors unless the status_code=True flag is enabled (then it adds status_code). Update the docs to reference the correct field(s) and when they’re present to avoid confusing users.
| print(result.get("status_code")) # HTTP status code if applicable | |
| print(result.get("http_status")) # HTTP status code for HTTP errors | |
| # If configured with status_code=True, errors also include "status_code" (same as http_status): | |
| # print(result.get("status_code")) |
| ] | ||
| keywords = ["clash", "clans", "api", "wrapper", "supercell"] | ||
| dependencies = ["httpx>=0.23.0,<1.0.0"] | ||
| dependencies = ["httpx>=0.23.0"] |
There was a problem hiding this comment.
httpx dependency was changed to remove the upper bound. If httpx introduces a breaking major release, this package could become installable with an incompatible version. Consider keeping an upper bound (or adding CI against newest httpx) to reduce the risk of runtime breaks for users.
| # Build URL | ||
| url = build_url(self.config.base_url, endpoint, params) | ||
|
|
||
| # Apply rate limiting | ||
| if self._rate_limiter: | ||
| await self._rate_limiter.acquire() | ||
|
|
||
| # Apply request middleware | ||
| headers = self.headers.copy() | ||
| url, headers, params = self.middleware.apply_request_middleware( | ||
| url, headers, params or {} | ||
| ) |
There was a problem hiding this comment.
In async POST, the URL is built from params before middleware runs. If request middleware modifies params (not url), the final request still uses the old query string because url is not rebuilt after apply_request_middleware(). Consider running middleware before build_url() or rebuilding the URL from the returned params so middleware can reliably influence query parameters.
| 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) |
There was a problem hiding this comment.
paginate() introduces new sync/async pagination behavior and relies on extract_items/extract_after_cursor handling dict vs model responses correctly. There don’t appear to be tests covering pagination (including cursor following, stopping conditions, and async mode), so regressions here may go unnoticed. Adding unit tests for both dict responses and Pydantic-enabled responses would help.
| def _add_status_code( | ||
| self, response: Dict[str, Any], status_code: int | ||
| ) -> Dict[str, Any]: | ||
| self, response: dict[str, Any], status_code: int | ||
| ) -> dict[str, Any]: | ||
| """Add status code to response if requested (backward compatibility)""" | ||
| if self.status_code: | ||
| response["status_code"] = status_code | ||
| return response |
There was a problem hiding this comment.
When use_dynamic_model/use_pydantic_models is enabled, create_dynamic_model() can return a Pydantic model instance (not a dict). _add_status_code() unconditionally assigns response["status_code"], which will raise for model objects. This also affects the cache-hit path where status_code is set on the already-converted response; consider guarding for non-dicts (skip, convert via model_dump(), or attach metadata in a supported way) so status_code=True doesn’t break model mode.
| def _save_cached_keys( | ||
| cache_path: Path, key_name: str, tokens: list[str], ip: str | ||
| ) -> None: | ||
| """Save tokens to disk, merging with any existing entries.""" | ||
| try: | ||
| cache_path.parent.mkdir(parents=True, exist_ok=True) | ||
|
|
||
| # Merge with existing data (other key_names) | ||
| data: dict[str, Any] = {} | ||
| if cache_path.is_file(): | ||
| try: | ||
| data = json.loads(cache_path.read_text()) | ||
| except (json.JSONDecodeError, OSError): | ||
| data = {} | ||
|
|
||
| data[key_name] = { | ||
| "tokens": tokens, | ||
| "ip": ip, | ||
| "created_at": datetime.now(timezone.utc).isoformat(), | ||
| } | ||
| cache_path.write_text(json.dumps(data, indent=2)) | ||
| logger.info("Saved %d token(s) to %s", len(tokens), cache_path) |
There was a problem hiding this comment.
Persisted API tokens are written to disk via write_text() without setting restrictive file permissions. On multi-user systems this file could become readable by other users depending on umask. Consider writing atomically and enforcing 0600 permissions (or equivalent) for the key cache file to reduce token exposure risk.
| use_dynamic_model, | ||
| _refresh_attempted=True, | ||
| ) | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| use_dynamic_model, | ||
| _refresh_attempted=True, | ||
| ) | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| use_dynamic_model, | ||
| _refresh_attempted=True, | ||
| ) | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| except Exception: | |
| # Intentionally ignore any errors while attempting to parse the | |
| # error body or refresh the token; in these cases we fall back | |
| # to the standard HTTP error handling logic below. |
| except Exception: | ||
| pass |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| pass | |
| except Exception as exc: | |
| logging.debug( | |
| "Failed to parse response body while checking for " | |
| "accessDenied.invalidIp during auto-refresh: %s", | |
| exc, | |
| ) |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 14 out of 17 changed files in this pull request and generated 15 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| tester = EndpointTester.__new__(EndpointTester) | ||
| tester.api = api | ||
| tester.clan_tag = clan_tag | ||
| tester.player_tag = player_tag | ||
| tester.results = {} |
There was a problem hiding this comment.
Using EndpointTester.__new__ bypasses __init__, which is fragile if initialization logic changes. Consider refactoring EndpointTester to accept an already-constructed CocApi (or adding an alternate constructor) instead of manual attribute setup.
| params: dict[str, Any] = {"limit": limit} | ||
| while True: | ||
| result = method(*args, params) | ||
| items = extract_items(result) |
There was a problem hiding this comment.
paginate() passes the params dict as a positional argument (method(*args, params)), which breaks for methods where params isn’t the last positional parameter (e.g. ApiMethods.clan(name, limit, params=...)). Pass params by keyword (and consider validating that the target callable accepts it) to avoid mis-binding.
| params: dict[str, Any] = {"limit": limit} | ||
| while True: | ||
| result = await method(*args, params) | ||
| items = extract_items(result) |
There was a problem hiding this comment.
Same issue as sync pagination: _apaginate passes the params dict positionally (await method(*args, params)), which can bind to the wrong parameter for some endpoints. Use await method(*args, params=params) (or signature-aware calling).
| # 3. Persisted key from a previous login | ||
| cached = _load_cached_keys(_DEFAULT_KEY_STORAGE_PATH, "cocapi_auto") | ||
| if cached: | ||
| tokens, _ip = cached | ||
| try: | ||
| return CocApi(tokens[0]) |
There was a problem hiding this comment.
When falling back to the persisted key, the cached IP is ignored and the token is used without validation. After an IP change this will lead to confusing 403 accessDenied.invalidIp errors; consider checking cached IP vs current detected IP (or doing a lightweight api.test() and falling back to re-login on invalidIp).
| "created_at": datetime.now(timezone.utc).isoformat(), | ||
| } | ||
| cache_path.write_text(json.dumps(data, indent=2)) | ||
| logger.info("Saved %d token(s) to %s", len(tokens), cache_path) | ||
| except OSError as e: |
There was a problem hiding this comment.
This writes API tokens to disk but doesn’t restrict file permissions. Consider setting the cache file to user-only permissions (e.g. 0o600 on POSIX) after writing to reduce the risk of token disclosure.
| data = resp.json() | ||
|
|
There was a problem hiding this comment.
_create_key() assumes JSON parsing succeeds and doesn’t validate resp.status_code. For non-200/non-JSON responses, resp.json() can raise and skip your KeyManagerError path; consider checking status and wrapping JSON decode failures consistently.
| data = resp.json() | |
| if resp.status_code != 200: | |
| raise KeyManagerError( | |
| f"Failed to create key: HTTP {resp.status_code}" | |
| ) | |
| try: | |
| data = resp.json() | |
| except ValueError as exc: | |
| raise KeyManagerError( | |
| "Failed to create key: invalid JSON in response" | |
| ) from exc |
| use_dynamic_model, | ||
| _refresh_attempted=True, | ||
| ) | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| except Exception: | |
| # If the error body cannot be parsed or inspected, skip auto-refresh | |
| # and fall back to the standard HTTP error handling below. |
| except Exception: | ||
| pass |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| pass | |
| except Exception as e: | |
| # Failed to inspect error body for auto-refresh; proceed with normal HTTP error handling. | |
| logging.getLogger(__name__).debug( | |
| "Failed to parse 403 response body for auto-refresh: %r", e | |
| ) |
| use_dynamic_model, | ||
| _refresh_attempted=True, | ||
| ) | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| except Exception: | |
| # If the response body is not valid JSON or does not contain the | |
| # expected "reason" field, fall back to the generic HTTP error | |
| # handling below without attempting auto-refresh. |
| except ImportError: | ||
| print( | ||
| "The cocapi CLI requires typer. Install it with:\n pip install 'cocapi[cli]'" | ||
| ) | ||
| sys.exit(1) | ||
|
|
There was a problem hiding this comment.
Print statement may execute during import.
| except ImportError: | |
| print( | |
| "The cocapi CLI requires typer. Install it with:\n pip install 'cocapi[cli]'" | |
| ) | |
| sys.exit(1) | |
| except ImportError as exc: | |
| raise ImportError( | |
| "The cocapi CLI requires typer. Install it with:\n pip install 'cocapi[cli]'" | |
| ) from exc |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 34 out of 37 changed files in this pull request and generated 13 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def set_key_manager_state( | ||
| self, | ||
| email: str, | ||
| password: str, | ||
| key_name: str, | ||
| key_count: int, | ||
| auto_refresh: bool, | ||
| persist_keys: bool = False, | ||
| key_storage_path: str | None = None, | ||
| ) -> None: | ||
| """Store key manager credentials for auto-refresh on 403.""" | ||
| self._km_email = email | ||
| self._km_password = password | ||
| self._km_key_name = key_name | ||
| self._km_key_count = key_count | ||
| self._km_auto_refresh = auto_refresh | ||
| self._km_persist_keys = persist_keys | ||
| self._km_key_storage_path = key_storage_path | ||
|
|
||
| def _should_auto_refresh_keys(self) -> bool: | ||
| """Check if auto key refresh is available and enabled.""" | ||
| return self._km_email is not None and self._km_auto_refresh | ||
|
|
||
| async def _refresh_token(self) -> bool: | ||
| """Refresh API token via AsyncKeyManager. Returns True if successful.""" | ||
| from .key_manager import AsyncKeyManager | ||
|
|
||
| try: | ||
| async with AsyncKeyManager( | ||
| email=self._km_email, # type: ignore[arg-type] | ||
| password=self._km_password, # type: ignore[arg-type] | ||
| key_name=self._km_key_name or "cocapi_auto", | ||
| key_count=self._km_key_count, | ||
| persist_keys=self._km_persist_keys, | ||
| key_storage_path=self._km_key_storage_path, | ||
| ) as km: |
There was a problem hiding this comment.
Async token refresh via AsyncKeyManager doesn’t pass through ApiConfig.key_description (or store it in set_key_manager_state), so async auto-refresh may create keys with the default description even when a custom description is configured. To keep sync/async behavior consistent, carry key_description through set_key_manager_state and into AsyncKeyManager(...).
| params: dict[str, Any] = {"limit": limit} | ||
| while True: | ||
| result = method(*args, params) | ||
| items = extract_items(result) | ||
| if not items: | ||
| break | ||
| yield from items | ||
| after = extract_after_cursor(result) | ||
| if not after: | ||
| break | ||
| params = {"limit": limit, "after": after} | ||
|
|
||
| async def _apaginate( | ||
| self, | ||
| method: Callable[..., Any], | ||
| *args: Any, | ||
| limit: int = 100, | ||
| ) -> Any: | ||
| """Asynchronous pagination generator.""" | ||
| params: dict[str, Any] = {"limit": limit} | ||
| while True: | ||
| result = await method(*args, params) | ||
| items = extract_items(result) | ||
| if not items: | ||
| break | ||
| for item in items: | ||
| yield item | ||
| after = extract_after_cursor(result) | ||
| if not after: | ||
| break | ||
| params = {"limit": limit, "after": after} |
There was a problem hiding this comment.
paginate() passes the params dict as a positional argument (method(*args, params) / await method(*args, params)), which breaks endpoints whose 2nd positional parameter is not params (e.g. ApiMethods.clan(name, limit, params=...) will receive a dict for limit). Pass params by keyword instead so pagination works across all list endpoints that accept a params kwarg.
| raise RuntimeError( | ||
| "EventStream requires CocApi in async mode. " | ||
| "Use 'async with CocApi(token) as api:'" | ||
| ) |
There was a problem hiding this comment.
EventStream only checks api.async_mode, but a user can set async_mode=True without entering async with (so _async_core is still None), which will later crash watchers with a RuntimeError. Consider validating that the API is actually in an active async context (e.g., _async_core is initialized) and raise a clear error here.
| ) | |
| ) | |
| # Defensive check: ensure the API is actually in an active async context. | |
| # Users might incorrectly set `async_mode=True` without using `async with`, | |
| # which leaves the internal async core uninitialized and would cause | |
| # runtime errors in watchers later. | |
| if getattr(api, "_async_core", None) is None: | |
| raise RuntimeError( | |
| "EventStream requires CocApi to be used inside an active async " | |
| "context. Make sure you create EventStream inside " | |
| "'async with CocApi(token) as api:'." | |
| ) |
| tester = EndpointTester.__new__(EndpointTester) | ||
| tester.api = api | ||
| tester.clan_tag = clan_tag | ||
| tester.player_tag = player_tag | ||
| tester.results = {} | ||
| tester.ok = 0 | ||
| tester.errors = 0 | ||
| tester.skipped = 0 |
There was a problem hiding this comment.
This uses EndpointTester.__new__ and then manually assigns fields, bypassing EndpointTester.__init__. That’s brittle (future init changes won’t be applied) and can lead to partially-initialized instances. Prefer updating EndpointTester.__init__ to accept a CocApi instance (or add an alternate constructor) instead of bypassing initialization.
| # KeyManager settings (used with CocApi.from_credentials()) | ||
| key_name: str = "cocapi_auto" # Name prefix for managed keys | ||
| key_count: int = 1 # Number of keys to maintain | ||
| key_description: str = "Auto-generated by cocapi KeyManager" | ||
| auto_refresh_keys: bool = True # Auto-refresh on accessDenied.invalidIp | ||
| persist_keys: bool = False # Save API keys locally for reuse across runs | ||
| key_storage_path: str | None = None # Custom path (default: ~/.cocapi/keys.json) |
There was a problem hiding this comment.
key_name is treated as an exact key name match in the key manager (key.get('name') != key_name), not a prefix. The comment here (“Name prefix for managed keys”) is misleading; consider updating it to reflect exact matching (or update the key manager to actually support prefix matching).
|
|
||
| from cocapi import CocApi | ||
| from cocapi.events._state import PollingState | ||
| from cocapi.events._types import Event, EventType |
There was a problem hiding this comment.
Import of 'Event' is not used.
| """ | ||
|
|
||
| import pytest | ||
| from unittest.mock import patch, AsyncMock |
There was a problem hiding this comment.
Import of 'AsyncMock' is not used.
| import pytest | ||
| from unittest.mock import patch, AsyncMock | ||
|
|
||
| from cocapi import CocApi, ApiConfig |
There was a problem hiding this comment.
Import of 'ApiConfig' is not used.
| try: | ||
| await self._task | ||
| except asyncio.CancelledError: | ||
| pass |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| pass | |
| logger.debug( | |
| "Watcher task %s cancelled during stop; this is expected.", | |
| self.__class__.__name__, | |
| ) |
| except Exception: | ||
| pass |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| pass | |
| except Exception as parse_error: | |
| logging.getLogger(__name__).debug( | |
| "Failed to parse error response JSON during auto-refresh handling: %s", | |
| parse_error, | |
| ) |
| "ip": ip, | ||
| "created_at": datetime.now(timezone.utc).isoformat(), | ||
| } | ||
| cache_path.write_text(json.dumps(data, indent=2)) |
Check failure
Code scanning / SonarCloud
I/O function calls should not be vulnerable to path injection attacks High
|
| data = json.loads(cache_path.read_text()) | ||
| if key_name in data: | ||
| del data[key_name] | ||
| cache_path.write_text(json.dumps(data, indent=2)) |
Check failure
Code scanning / SonarCloud
I/O function calls should not be vulnerable to path injection attacks High




No description provided.