diff --git a/awsysco/__init__.py b/awsysco/__init__.py index 23676bd..6b36d82 100644 --- a/awsysco/__init__.py +++ b/awsysco/__init__.py @@ -1,6 +1,6 @@ """AWSYS.CO Python SDK — Official client library for the AWSYS.CO URL Shortener API.""" -from .client import Client +from .client import AsyncClient, Client from .exceptions import ( AwsysAuthError, AwsysConflictError, @@ -11,21 +11,35 @@ AwsysValidationError, ) from .models import ( + AffiliateProgram, BulkLinkResult, BulkResult, ClickEvent, + CustomDomain, Folder, FolderList, + GeoRestriction, Link, LinkList, LinkStats, MeResponse, + NamespaceCheckResult, + NamespaceInfo, + OgMeta, + QRSettings, + RoutingRule, + SavedView, + SavedViewFilters, + TrustScoreResult, + UtmTemplate, + Webhook, ) -__version__ = "0.1.0" +__version__ = "1.0.0" __all__ = [ - # Client + # Clients "Client", + "AsyncClient", # Exceptions "AwsysError", "AwsysAuthError", @@ -34,7 +48,7 @@ "AwsysConflictError", "AwsysValidationError", "AwsysRateLimitError", - # Models + # Core models "Link", "LinkList", "LinkStats", @@ -44,4 +58,26 @@ "BulkResult", "BulkLinkResult", "MeResponse", + # Links advanced models + "RoutingRule", + "OgMeta", + "GeoRestriction", + # QR + "QRSettings", + # Trust Score + "TrustScoreResult", + # Namespace + "NamespaceInfo", + "NamespaceCheckResult", + # UTM Templates + "UtmTemplate", + # Webhooks + "Webhook", + # Saved Views + "SavedViewFilters", + "SavedView", + # Custom Domains + "CustomDomain", + # Affiliate + "AffiliateProgram", ] diff --git a/awsysco/_async_http.py b/awsysco/_async_http.py new file mode 100644 index 0000000..e271646 --- /dev/null +++ b/awsysco/_async_http.py @@ -0,0 +1,158 @@ +"""Async HTTP client wrapper with retry logic and error mapping.""" + +from __future__ import annotations + +import asyncio +from typing import Any, Dict, Optional + +import httpx + +from .exceptions import ( + AwsysAuthError, + AwsysConflictError, + AwsysError, + AwsysForbiddenError, + AwsysNotFoundError, + AwsysRateLimitError, + AwsysValidationError, +) + +_MAX_RETRIES = 3 +_RETRY_BASE_DELAY = 1.0 # seconds + + +def _parse_error(response: httpx.Response) -> AwsysError: + """Parse an HTTP error response and return the appropriate exception.""" + status = response.status_code + raw: Any = None + message: str = f"HTTP {status}" + code: Optional[str] = None + + try: + data = response.json() + raw = data + msg_field = data.get("message") + if msg_field and isinstance(msg_field, str): + message = msg_field + elif isinstance(data.get("error"), str): + message = data["error"] + code = data.get("code") + except Exception: + raw = response.text + if raw: + message = raw + + kwargs: Dict[str, Any] = {"code": code, "status": status, "raw": raw} + + if status == 400: + return AwsysValidationError(message, **kwargs) + if status == 401: + return AwsysAuthError(message, **kwargs) + if status == 403: + return AwsysForbiddenError(message, **kwargs) + if status == 404: + return AwsysNotFoundError(message, **kwargs) + if status == 409: + return AwsysConflictError(message, **kwargs) + if status == 429: + retry_after: Optional[float] = None + ra_header = response.headers.get("Retry-After") + if ra_header: + try: + retry_after = float(ra_header) + except ValueError: + pass + return AwsysRateLimitError(message, retry_after=retry_after, **kwargs) + + return AwsysError(message, **kwargs) + + +class AsyncHttpClient: + """Async wrapper around httpx.AsyncClient with auth, retries, and error mapping.""" + + def __init__(self, api_key: str, base_url: str, timeout: float = 30.0) -> None: + self._base_url = base_url.rstrip("/") + self._client = httpx.AsyncClient( + base_url=self._base_url, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": "awsysco-python-sdk/1.0.0", + }, + timeout=timeout, + ) + + async def _request( + self, + method: str, + path: str, + *, + params: Optional[Dict[str, Any]] = None, + json: Optional[Any] = None, + ) -> Any: + """Execute an async HTTP request with 429-retry logic.""" + attempt = 0 + while True: + response = await self._client.request(method, path, params=params, json=json) + + if response.status_code == 429 and attempt < _MAX_RETRIES: + exc = _parse_error(response) + assert isinstance(exc, AwsysRateLimitError) + delay = exc.retry_after or (_RETRY_BASE_DELAY * (2 ** attempt)) + await asyncio.sleep(delay) + attempt += 1 + continue + + if response.is_error: + raise _parse_error(response) + + # 204 No Content + if response.status_code == 204 or not response.content: + return None + + return response.json() + + async def get(self, path: str, *, params: Optional[Dict[str, Any]] = None) -> Any: + return await self._request("GET", path, params=params) + + async def get_text(self, path: str, *, params: Optional[Dict[str, Any]] = None) -> str: + """Like get() but returns response.text instead of response.json().""" + attempt = 0 + while True: + response = await self._client.request("GET", path, params=params) + + if response.status_code == 429 and attempt < _MAX_RETRIES: + exc = _parse_error(response) + assert isinstance(exc, AwsysRateLimitError) + delay = exc.retry_after or (_RETRY_BASE_DELAY * (2 ** attempt)) + await asyncio.sleep(delay) + attempt += 1 + continue + + if response.is_error: + raise _parse_error(response) + + return response.text + + async def post(self, path: str, *, json: Optional[Any] = None) -> Any: + return await self._request("POST", path, json=json) + + async def patch(self, path: str, *, json: Optional[Any] = None) -> Any: + return await self._request("PATCH", path, json=json) + + async def put(self, path: str, *, json: Optional[Any] = None) -> Any: + return await self._request("PUT", path, json=json) + + async def delete(self, path: str) -> Any: + return await self._request("DELETE", path) + + async def aclose(self) -> None: + """Close the underlying async HTTP connection pool.""" + await self._client.aclose() + + async def __aenter__(self) -> "AsyncHttpClient": + return self + + async def __aexit__(self, *args: Any) -> None: + await self.aclose() diff --git a/awsysco/_http.py b/awsysco/_http.py index f9a9099..70c2f2d 100644 --- a/awsysco/_http.py +++ b/awsysco/_http.py @@ -80,7 +80,7 @@ def __init__(self, api_key: str, base_url: str, timeout: float = 30.0) -> None: "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Accept": "application/json", - "User-Agent": "awsysco-python-sdk/0.1.0", + "User-Agent": "awsysco-python-sdk/1.0.0", }, timeout=timeout, ) @@ -118,12 +118,34 @@ def _request( def get(self, path: str, *, params: Optional[Dict[str, Any]] = None) -> Any: return self._request("GET", path, params=params) + def get_text(self, path: str, *, params: Optional[Dict[str, Any]] = None) -> str: + """Like get() but returns response.text instead of response.json().""" + attempt = 0 + while True: + response = self._client.request("GET", path, params=params) + + if response.status_code == 429 and attempt < _MAX_RETRIES: + exc = _parse_error(response) + assert isinstance(exc, AwsysRateLimitError) + delay = exc.retry_after or (_RETRY_BASE_DELAY * (2 ** attempt)) + time.sleep(delay) + attempt += 1 + continue + + if response.is_error: + raise _parse_error(response) + + return response.text + def post(self, path: str, *, json: Optional[Any] = None) -> Any: return self._request("POST", path, json=json) def patch(self, path: str, *, json: Optional[Any] = None) -> Any: return self._request("PATCH", path, json=json) + def put(self, path: str, *, json: Optional[Any] = None) -> Any: + return self._request("PUT", path, json=json) + def delete(self, path: str) -> Any: return self._request("DELETE", path) diff --git a/awsysco/async_resources/__init__.py b/awsysco/async_resources/__init__.py new file mode 100644 index 0000000..395d5cf --- /dev/null +++ b/awsysco/async_resources/__init__.py @@ -0,0 +1 @@ +"""Async resource classes for the AWSYS.CO SDK.""" diff --git a/awsysco/async_resources/affiliate.py b/awsysco/async_resources/affiliate.py new file mode 100644 index 0000000..8d821c3 --- /dev/null +++ b/awsysco/async_resources/affiliate.py @@ -0,0 +1,80 @@ +"""Async Affiliate resource.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .._async_http import AsyncHttpClient +from ..models import AffiliateProgram + + +class AsyncAffiliateResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def create_program(self, name: str, commission_type: str, **kwargs: Any) -> AffiliateProgram: + body: Dict[str, Any] = {"name": name, "commissionType": commission_type} + field_map = {"description": "description", "cpc_rate": "cpcRate", "cpa_rate": "cpaRate", "cookie_days": "cookieDays"} + for k, v in kwargs.items(): + body[field_map.get(k, k)] = v + data = await self._http.post("/api/affiliate/programs", json=body) + return AffiliateProgram.model_validate(data) + + async def list_programs(self) -> List[AffiliateProgram]: + data = await self._http.get("/api/affiliate/programs") + if isinstance(data, list): + return [AffiliateProgram.model_validate(item) for item in data] + items = data.get("programs", []) if isinstance(data, dict) else [] + return [AffiliateProgram.model_validate(item) for item in items] + + async def get_program(self, program_id: str) -> AffiliateProgram: + data = await self._http.get(f"/api/affiliate/programs/{program_id}") + return AffiliateProgram.model_validate(data) + + async def update_program(self, program_id: str, **kwargs: Any) -> AffiliateProgram: + body: Dict[str, Any] = {} + field_map = {"cpc_rate": "cpcRate", "cpa_rate": "cpaRate", "cookie_days": "cookieDays", "commission_type": "commissionType"} + for k, v in kwargs.items(): + body[field_map.get(k, k)] = v + data = await self._http.patch(f"/api/affiliate/programs/{program_id}", json=body) + return AffiliateProgram.model_validate(data) + + async def get_program_stats(self, program_id: str, *, period: str = "30d") -> dict: + return await self._http.get(f"/api/affiliate/programs/{program_id}/stats", params={"period": period}) or {} + + async def list_partners(self, program_id: str) -> List[dict]: + data = await self._http.get(f"/api/affiliate/programs/{program_id}/partners") + if isinstance(data, list): + return data + return data.get("partners", []) if isinstance(data, dict) else [] + + async def update_partner_status(self, program_id: str, partner_id: str, status: str) -> dict: + return await self._http.patch(f"/api/affiliate/programs/{program_id}/partners/{partner_id}", json={"status": status}) or {} + + async def discover(self, *, limit: int = 20) -> List[AffiliateProgram]: + data = await self._http.get("/api/affiliate/discover", params={"limit": limit}) + if isinstance(data, list): + return [AffiliateProgram.model_validate(item) for item in data] + items = data.get("programs", []) if isinstance(data, dict) else [] + return [AffiliateProgram.model_validate(item) for item in items] + + async def join(self, program_id: str, *, partner_code: Optional[str] = None) -> dict: + body: Dict[str, Any] = {} + if partner_code is not None: + body["partnerCode"] = partner_code + return await self._http.post(f"/api/affiliate/join/{program_id}", json=body) or {} + + async def list_partnerships(self) -> List[dict]: + data = await self._http.get("/api/affiliate/partnerships") + if isinstance(data, list): + return data + return data.get("partnerships", []) if isinstance(data, dict) else [] + + async def get_partnership_stats(self, partnership_id: str, *, period: str = "30d") -> dict: + return await self._http.get(f"/api/affiliate/partnerships/{partnership_id}/stats", params={"period": period}) or {} + + async def leave_program(self, partnership_id: str) -> dict: + return await self._http.delete(f"/api/affiliate/partnerships/{partnership_id}") or {} + + async def get_limits(self) -> dict: + return await self._http.get("/api/affiliate/limits") or {} diff --git a/awsysco/async_resources/agentlink.py b/awsysco/async_resources/agentlink.py new file mode 100644 index 0000000..7222cbf --- /dev/null +++ b/awsysco/async_resources/agentlink.py @@ -0,0 +1,19 @@ +"""Async AgentLink resource.""" + +from __future__ import annotations + +from .._async_http import AsyncHttpClient + + +class AsyncAgentlinkResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def subscribe(self, email: str) -> dict: + return await self._http.post("/api/agentlink/subscribe", json={"email": email}) or {} + + async def get_link_stats(self, short_path: str, *, period_days: int = 7) -> dict: + return await self._http.get(f"/api/agentlink/links/{short_path}/stats", params={"period": period_days}) or {} + + async def get_account_stats(self, *, period_days: int = 7) -> dict: + return await self._http.get("/api/agentlink/account/stats", params={"period": period_days}) or {} diff --git a/awsysco/async_resources/analytics.py b/awsysco/async_resources/analytics.py new file mode 100644 index 0000000..592efd7 --- /dev/null +++ b/awsysco/async_resources/analytics.py @@ -0,0 +1,36 @@ +"""Async Analytics resource.""" + +from __future__ import annotations + +from typing import List, Optional + +from .._async_http import AsyncHttpClient +from ..models import ClickEvent, LinkStats + + +class AsyncAnalyticsResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def get_stats(self, short_path: str, *, period: Optional[str] = None) -> LinkStats: + params = {} + if period is not None: + params["period"] = period + data = await self._http.get( + f"/api/v1/links/{short_path}/stats", + params=params if params else None, + ) + return LinkStats.model_validate(data) + + async def get_recent_clicks(self, *, limit: Optional[int] = None) -> List[ClickEvent]: + params = {} + if limit is not None: + params["limit"] = limit + data = await self._http.get( + "/api/user/recent-clicks", + params=params if params else None, + ) + if isinstance(data, list): + return [ClickEvent.model_validate(item) for item in data] + items = data.get("clicks", data.get("recentClicks", [])) + return [ClickEvent.model_validate(item) for item in items] diff --git a/awsysco/async_resources/bulk.py b/awsysco/async_resources/bulk.py new file mode 100644 index 0000000..a3fa55c --- /dev/null +++ b/awsysco/async_resources/bulk.py @@ -0,0 +1,33 @@ +"""Async Bulk resource.""" + +from __future__ import annotations + +from typing import Any, Dict, List + +from .._async_http import AsyncHttpClient +from ..models import BulkResult + + +class AsyncBulkResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def create(self, urls: List[Dict[str, Any]]) -> BulkResult: + payload: List[Dict[str, Any]] = [] + for item in urls: + entry: Dict[str, Any] = {"url": item["url"]} + if "custom_slug" in item: + entry["customSlug"] = item["custom_slug"] + if "customSlug" in item: + entry["customSlug"] = item["customSlug"] + if "expires_at" in item: + entry["expiresAt"] = item["expires_at"] + if "expiresAt" in item: + entry["expiresAt"] = item["expiresAt"] + if "max_clicks" in item: + entry["maxClicks"] = item["max_clicks"] + if "maxClicks" in item: + entry["maxClicks"] = item["maxClicks"] + payload.append(entry) + data = await self._http.post("/api/v1/bulk", json={"urls": payload}) + return BulkResult.model_validate(data) diff --git a/awsysco/async_resources/custom_domains.py b/awsysco/async_resources/custom_domains.py new file mode 100644 index 0000000..9ea55f2 --- /dev/null +++ b/awsysco/async_resources/custom_domains.py @@ -0,0 +1,41 @@ +"""Async Custom Domains resource.""" + +from __future__ import annotations + +from typing import Any, Dict, Optional + +from .._async_http import AsyncHttpClient +from ..models import CustomDomain + + +class AsyncCustomDomainsResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def list(self) -> dict: + return await self._http.get("/api/user/domains") or {} + + async def add(self, domain: str) -> dict: + return await self._http.post("/api/user/domains", json={"domain": domain}) or {} + + async def verify(self, domain: str) -> dict: + return await self._http.get(f"/api/user/domains/{domain}/verify") or {} + + async def activate(self, domain: str) -> CustomDomain: + data = await self._http.post(f"/api/user/domains/{domain}/activate") + return CustomDomain.model_validate(data) + + async def update(self, domain: str, *, is_default: Optional[bool] = None, not_found_html: Optional[str] = None) -> CustomDomain: + body: Dict[str, Any] = {} + if is_default is not None: + body["isDefault"] = is_default + if not_found_html is not None: + body["notFoundHtml"] = not_found_html + data = await self._http.patch(f"/api/user/domains/{domain}", json=body) + return CustomDomain.model_validate(data) + + async def remove(self, domain: str) -> dict: + return await self._http.delete(f"/api/user/domains/{domain}") or {} + + async def check(self, hostname: str) -> dict: + return await self._http.get(f"/api/domains/check/{hostname}") or {} diff --git a/awsysco/async_resources/data_export.py b/awsysco/async_resources/data_export.py new file mode 100644 index 0000000..022a0b7 --- /dev/null +++ b/awsysco/async_resources/data_export.py @@ -0,0 +1,16 @@ +"""Async Data Export resource.""" + +from __future__ import annotations + +from .._async_http import AsyncHttpClient + + +class AsyncDataExportResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def export_links(self) -> str: + return await self._http.get_text("/api/export/links") + + async def export_link_stats(self, short_path: str) -> str: + return await self._http.get_text(f"/api/export/stats/{short_path}") diff --git a/awsysco/async_resources/folders.py b/awsysco/async_resources/folders.py new file mode 100644 index 0000000..9a4bc44 --- /dev/null +++ b/awsysco/async_resources/folders.py @@ -0,0 +1,42 @@ +"""Async Folders resource.""" + +from __future__ import annotations + +from typing import Any, Dict, Optional + +from .._async_http import AsyncHttpClient +from ..models import Folder, FolderList + + +class AsyncFoldersResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def list(self) -> FolderList: + data = await self._http.get("/api/v1/folders") + return FolderList.model_validate(data) + + async def create(self, name: str, *, color: Optional[str] = None) -> Folder: + body: Dict[str, Any] = {"name": name} + if color is not None: + body["color"] = color + data = await self._http.post("/api/v1/folders", json=body) + return Folder.model_validate(data) + + async def update(self, folder_id: str, *, name: Optional[str] = None, color: Optional[str] = None) -> Folder: + body: Dict[str, Any] = {} + if name is not None: + body["name"] = name + if color is not None: + body["color"] = color + data = await self._http.patch(f"/api/v1/folders/{folder_id}", json=body) + return Folder.model_validate(data) + + async def delete(self, folder_id: str) -> None: + await self._http.delete(f"/api/v1/folders/{folder_id}") + + async def assign_link(self, short_path: str, folder_id: str) -> None: + await self._http.post(f"/api/v1/links/{short_path}/folder", json={"folderId": folder_id}) + + async def remove_link(self, short_path: str) -> None: + await self._http.post(f"/api/v1/links/{short_path}/folder", json={"folderId": None}) diff --git a/awsysco/async_resources/links.py b/awsysco/async_resources/links.py new file mode 100644 index 0000000..02ce8ae --- /dev/null +++ b/awsysco/async_resources/links.py @@ -0,0 +1,102 @@ +"""Async Links resource.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .._async_http import AsyncHttpClient +from ..models import Link, LinkList + + +class AsyncLinksResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def create( + self, + url: str, + *, + custom_slug: Optional[str] = None, + expires_at: Optional[str] = None, + max_clicks: Optional[int] = None, + routing_rules: Optional[List[Dict[str, str]]] = None, + og_meta: Optional[Dict[str, str]] = None, + geo_restriction: Optional[Dict[str, List[str]]] = None, + password: Optional[str] = None, + pass_ad_click_ids: Optional[bool] = None, + folder_id: Optional[str] = None, + tags: Optional[List[str]] = None, + ) -> Link: + body: Dict[str, Any] = {"url": url} + if custom_slug is not None: + body["customSlug"] = custom_slug + if expires_at is not None: + body["expiresAt"] = expires_at + if max_clicks is not None: + body["maxClicks"] = max_clicks + if routing_rules is not None: + body["routingRules"] = routing_rules + if og_meta is not None: + body["ogMeta"] = og_meta + if geo_restriction is not None: + body["geoRestriction"] = geo_restriction + if password is not None: + body["password"] = password + if pass_ad_click_ids is not None: + body["passAdClickIds"] = pass_ad_click_ids + if folder_id is not None: + body["folderId"] = folder_id + if tags is not None: + body["tags"] = tags + data = await self._http.post("/api/v1/links", json=body) + return Link.model_validate(data) + + async def list(self, *, limit: int = 20, offset: int = 0) -> LinkList: + data = await self._http.get("/api/v1/links", params={"limit": limit, "offset": offset}) + return LinkList.model_validate(data) + + async def get(self, short_path: str) -> Link: + data = await self._http.get(f"/api/v1/links/{short_path}") + return Link.model_validate(data) + + async def update( + self, + short_path: str, + *, + url: Optional[str] = None, + expires_at: Optional[str] = None, + max_clicks: Optional[int] = None, + routing_rules: Optional[List[Dict[str, str]]] = None, + og_meta: Optional[Dict[str, str]] = None, + geo_restriction: Optional[Dict[str, List[str]]] = None, + password: Optional[str] = None, + pass_ad_click_ids: Optional[bool] = None, + folder_id: Optional[str] = None, + tags: Optional[List[str]] = None, + ) -> Link: + body: Dict[str, Any] = {} + if url is not None: + body["url"] = url + if expires_at is not None: + body["expiresAt"] = expires_at + if max_clicks is not None: + body["maxClicks"] = max_clicks + if routing_rules is not None: + body["routingRules"] = routing_rules + if og_meta is not None: + body["ogMeta"] = og_meta + if geo_restriction is not None: + body["geoRestriction"] = geo_restriction + if password is not None: + body["password"] = password + if pass_ad_click_ids is not None: + body["passAdClickIds"] = pass_ad_click_ids + if folder_id is not None: + body["folderId"] = folder_id + if tags is not None: + body["tags"] = tags + data = await self._http.patch(f"/api/v1/links/{short_path}", json=body) + return Link.model_validate(data) + + async def delete(self, short_path: str) -> None: + await self._http.delete(f"/api/v1/links/{short_path}") diff --git a/awsysco/async_resources/me.py b/awsysco/async_resources/me.py new file mode 100644 index 0000000..63bf597 --- /dev/null +++ b/awsysco/async_resources/me.py @@ -0,0 +1,15 @@ +"""Async Me resource.""" + +from __future__ import annotations + +from .._async_http import AsyncHttpClient +from ..models import MeResponse + + +class AsyncMeResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def get(self) -> MeResponse: + data = await self._http.get("/api/v1/me") + return MeResponse.model_validate(data) diff --git a/awsysco/async_resources/namespace.py b/awsysco/async_resources/namespace.py new file mode 100644 index 0000000..dd9f355 --- /dev/null +++ b/awsysco/async_resources/namespace.py @@ -0,0 +1,26 @@ +"""Async Namespace resource.""" + +from __future__ import annotations + +from .._async_http import AsyncHttpClient +from ..models import NamespaceCheckResult, NamespaceInfo + + +class AsyncNamespaceResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def get(self) -> NamespaceInfo: + data = await self._http.get("/api/user/namespace") + return NamespaceInfo.model_validate(data) + + async def check(self, namespace: str) -> NamespaceCheckResult: + data = await self._http.get(f"/api/namespace/check/{namespace}") + return NamespaceCheckResult.model_validate(data) + + async def claim(self, namespace: str) -> NamespaceInfo: + data = await self._http.post("/api/user/namespace", json={"namespace": namespace}) + return NamespaceInfo.model_validate(data) + + async def release(self) -> dict: + return await self._http.delete("/api/user/namespace") or {} diff --git a/awsysco/async_resources/qr.py b/awsysco/async_resources/qr.py new file mode 100644 index 0000000..b91cd35 --- /dev/null +++ b/awsysco/async_resources/qr.py @@ -0,0 +1,27 @@ +"""Async QR resource.""" + +from __future__ import annotations + +from typing import Any, Dict +from urllib.parse import urlencode + +from .._async_http import AsyncHttpClient +from ..models import QRSettings + + +class AsyncQRResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + self._base_url = http._base_url + + def get_url(self, short_code: str, *, size: int = 300, color: str = "000000", bg_color: str = "FFFFFF") -> str: + params = urlencode({"size": size, "color": color, "bgColor": bg_color}) + return f"{self._base_url}/api/qr/{short_code}?{params}" + + async def get_settings(self, short_path: str) -> QRSettings: + data = await self._http.get(f"/api/link/{short_path}/qr-settings") + return QRSettings.model_validate(data) + + async def update_settings(self, short_path: str, settings: Dict[str, Any]) -> QRSettings: + data = await self._http.put(f"/api/link/{short_path}/qr-settings", json=settings) + return QRSettings.model_validate(data) diff --git a/awsysco/async_resources/saved_views.py b/awsysco/async_resources/saved_views.py new file mode 100644 index 0000000..77e2565 --- /dev/null +++ b/awsysco/async_resources/saved_views.py @@ -0,0 +1,36 @@ +"""Async Saved Views resource.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .._async_http import AsyncHttpClient +from ..models import SavedView + + +class AsyncSavedViewsResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def list(self) -> List[SavedView]: + data = await self._http.get("/api/views") + if isinstance(data, list): + return [SavedView.model_validate(item) for item in data] + items = data.get("views", []) if isinstance(data, dict) else [] + return [SavedView.model_validate(item) for item in items] + + async def create(self, name: str, filters: Dict[str, Any]) -> SavedView: + data = await self._http.post("/api/views", json={"name": name, "filters": filters}) + return SavedView.model_validate(data) + + async def update(self, view_id: str, *, name: Optional[str] = None, filters: Optional[Dict[str, Any]] = None) -> SavedView: + body: Dict[str, Any] = {} + if name is not None: + body["name"] = name + if filters is not None: + body["filters"] = filters + data = await self._http.patch(f"/api/views/{view_id}", json=body) + return SavedView.model_validate(data) + + async def delete(self, view_id: str) -> None: + await self._http.delete(f"/api/views/{view_id}") diff --git a/awsysco/async_resources/tags.py b/awsysco/async_resources/tags.py new file mode 100644 index 0000000..d354c68 --- /dev/null +++ b/awsysco/async_resources/tags.py @@ -0,0 +1,21 @@ +"""Async Tags resource.""" + +from __future__ import annotations + +from urllib.parse import quote + +from .._async_http import AsyncHttpClient + + +class AsyncTagsResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def add(self, short_path: str, tag: str) -> dict: + encoded = quote(short_path, safe="") + return await self._http.post(f"/api/link/{encoded}/tags", json={"tag": tag}) or {} + + async def remove(self, short_path: str, tag: str) -> dict: + encoded = quote(short_path, safe="") + encoded_tag = quote(tag, safe="") + return await self._http.delete(f"/api/link/{encoded}/tags/{encoded_tag}") or {} diff --git a/awsysco/async_resources/trust_score.py b/awsysco/async_resources/trust_score.py new file mode 100644 index 0000000..6c10e7e --- /dev/null +++ b/awsysco/async_resources/trust_score.py @@ -0,0 +1,18 @@ +"""Async Trust Score resource.""" + +from __future__ import annotations + +from urllib.parse import quote + +from .._async_http import AsyncHttpClient +from ..models import TrustScoreResult + + +class AsyncTrustScoreResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def scan(self, short_path: str) -> TrustScoreResult: + encoded = quote(short_path, safe="") + data = await self._http.get(f"/api/link-scan/{encoded}") + return TrustScoreResult.model_validate(data) diff --git a/awsysco/async_resources/utm_templates.py b/awsysco/async_resources/utm_templates.py new file mode 100644 index 0000000..0c42302 --- /dev/null +++ b/awsysco/async_resources/utm_templates.py @@ -0,0 +1,28 @@ +"""Async UTM Templates resource.""" + +from __future__ import annotations + +from typing import Any, Dict, List + +from .._async_http import AsyncHttpClient +from ..models import UtmTemplate + + +class AsyncUtmTemplatesResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def list(self) -> List[UtmTemplate]: + resp = await self._http.get("/api/v1/me") + items = resp.get("utmTemplates", []) if isinstance(resp, dict) else [] + return [UtmTemplate.model_validate(item) for item in items] + + async def create(self, name: str, source: str, medium: str, campaign: str, *, term: str = "", content: str = "") -> dict: + body: Dict[str, Any] = { + "name": name, "source": source, "medium": medium, + "campaign": campaign, "term": term, "content": content, + } + return await self._http.post("/api/user/utm-templates", json=body) or {} + + async def delete(self, template_id: str) -> dict: + return await self._http.delete(f"/api/user/utm-templates/{template_id}") or {} diff --git a/awsysco/async_resources/webhooks.py b/awsysco/async_resources/webhooks.py new file mode 100644 index 0000000..7d59d95 --- /dev/null +++ b/awsysco/async_resources/webhooks.py @@ -0,0 +1,42 @@ +"""Async Webhooks resource.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .._async_http import AsyncHttpClient +from ..models import Webhook + + +class AsyncWebhooksResource: + def __init__(self, http: AsyncHttpClient) -> None: + self._http = http + + async def list_event_types(self) -> dict: + return await self._http.get("/api/webhooks/event-types") or {} + + async def list(self) -> dict: + return await self._http.get("/api/webhooks") or {} + + async def create(self, url: str, events: List[str], *, name: Optional[str] = None, secret: Optional[str] = None) -> Webhook: + body: Dict[str, Any] = {"url": url, "events": events} + if name is not None: + body["name"] = name + if secret is not None: + body["secret"] = secret + data = await self._http.post("/api/webhooks", json=body) + return Webhook.model_validate(data) + + async def update(self, webhook_id: str, **kwargs: Any) -> Webhook: + body: Dict[str, Any] = {} + key_map = {"url": "url", "events": "events", "name": "name", "secret": "secret", "enabled": "enabled"} + for k, v in kwargs.items(): + body[key_map.get(k, k)] = v + data = await self._http.patch(f"/api/webhooks/{webhook_id}", json=body) + return Webhook.model_validate(data) + + async def delete(self, webhook_id: str) -> dict: + return await self._http.delete(f"/api/webhooks/{webhook_id}") or {} + + async def test(self, webhook_id: str, event_type: str) -> dict: + return await self._http.post(f"/api/webhooks/{webhook_id}/test", json={"eventType": event_type}) or {} diff --git a/awsysco/client.py b/awsysco/client.py index 8add2a7..73cc194 100644 --- a/awsysco/client.py +++ b/awsysco/client.py @@ -1,22 +1,49 @@ -"""AWSYS.CO API Client.""" +"""AWSYS.CO API Client — sync and async.""" from __future__ import annotations from typing import Optional +from ._async_http import AsyncHttpClient from ._http import HttpClient +from .async_resources.affiliate import AsyncAffiliateResource +from .async_resources.agentlink import AsyncAgentlinkResource +from .async_resources.analytics import AsyncAnalyticsResource +from .async_resources.bulk import AsyncBulkResource +from .async_resources.custom_domains import AsyncCustomDomainsResource +from .async_resources.data_export import AsyncDataExportResource +from .async_resources.folders import AsyncFoldersResource +from .async_resources.links import AsyncLinksResource +from .async_resources.me import AsyncMeResource +from .async_resources.namespace import AsyncNamespaceResource +from .async_resources.qr import AsyncQRResource +from .async_resources.saved_views import AsyncSavedViewsResource +from .async_resources.tags import AsyncTagsResource +from .async_resources.trust_score import AsyncTrustScoreResource +from .async_resources.utm_templates import AsyncUtmTemplatesResource +from .async_resources.webhooks import AsyncWebhooksResource +from .resources.affiliate import AffiliateResource +from .resources.agentlink import AgentlinkResource from .resources.analytics import AnalyticsResource from .resources.bulk import BulkResource +from .resources.custom_domains import CustomDomainsResource +from .resources.data_export import DataExportResource from .resources.folders import FoldersResource from .resources.links import LinksResource from .resources.me import MeResource +from .resources.namespace import NamespaceResource from .resources.qr import QRResource +from .resources.saved_views import SavedViewsResource +from .resources.tags import TagsResource +from .resources.trust_score import TrustScoreResource +from .resources.utm_templates import UtmTemplatesResource +from .resources.webhooks import WebhooksResource _DEFAULT_BASE_URL = "https://awsys.co" class Client: - """Top-level client for the AWSYS.CO API. + """Top-level synchronous client for the AWSYS.CO API. Example:: @@ -43,6 +70,7 @@ def __init__( ) -> None: self._http = HttpClient(api_key=api_key, base_url=base_url, timeout=timeout) + # Core resources self.links = LinksResource(self._http) self.analytics = AnalyticsResource(self._http) self.qr = QRResource(self._http) @@ -50,6 +78,20 @@ def __init__( self.bulk = BulkResource(self._http) self.me = MeResource(self._http) + # Phase 2 — simple new resources + self.tags = TagsResource(self._http) + self.trust_score = TrustScoreResource(self._http) + self.data_export = DataExportResource(self._http) + self.namespace = NamespaceResource(self._http) + self.utm_templates = UtmTemplatesResource(self._http) + + # Phase 3 — complex new resources + self.webhooks = WebhooksResource(self._http) + self.saved_views = SavedViewsResource(self._http) + self.custom_domains = CustomDomainsResource(self._http) + self.agentlink = AgentlinkResource(self._http) + self.affiliate = AffiliateResource(self._http) + def close(self) -> None: """Close the underlying HTTP connection pool.""" self._http.close() @@ -59,3 +101,62 @@ def __enter__(self) -> "Client": def __exit__(self, *args: object) -> None: self.close() + + +class AsyncClient: + """Top-level asynchronous client for the AWSYS.CO API. + + Use as an async context manager:: + + from awsysco import AsyncClient + + async with AsyncClient(api_key="awsys_...") as client: + link = await client.links.create("https://example.com") + print(link.short_url) + + Args: + api_key: Your AWSYS API key (starts with ``awsys_``). + base_url: API base URL. Defaults to ``https://awsys.co``. + timeout: HTTP request timeout in seconds (default 30). + """ + + def __init__( + self, + api_key: str, + *, + base_url: str = _DEFAULT_BASE_URL, + timeout: float = 30.0, + ) -> None: + self._http = AsyncHttpClient(api_key=api_key, base_url=base_url, timeout=timeout) + + # Core resources + self.links = AsyncLinksResource(self._http) + self.analytics = AsyncAnalyticsResource(self._http) + self.qr = AsyncQRResource(self._http) + self.folders = AsyncFoldersResource(self._http) + self.bulk = AsyncBulkResource(self._http) + self.me = AsyncMeResource(self._http) + + # Phase 2 — simple new resources + self.tags = AsyncTagsResource(self._http) + self.trust_score = AsyncTrustScoreResource(self._http) + self.data_export = AsyncDataExportResource(self._http) + self.namespace = AsyncNamespaceResource(self._http) + self.utm_templates = AsyncUtmTemplatesResource(self._http) + + # Phase 3 — complex new resources + self.webhooks = AsyncWebhooksResource(self._http) + self.saved_views = AsyncSavedViewsResource(self._http) + self.custom_domains = AsyncCustomDomainsResource(self._http) + self.agentlink = AsyncAgentlinkResource(self._http) + self.affiliate = AsyncAffiliateResource(self._http) + + async def aclose(self) -> None: + """Close the underlying async HTTP connection pool.""" + await self._http.aclose() + + async def __aenter__(self) -> "AsyncClient": + return self + + async def __aexit__(self, *args: object) -> None: + await self.aclose() diff --git a/awsysco/models.py b/awsysco/models.py index 1bd7c3b..cf5aa85 100644 --- a/awsysco/models.py +++ b/awsysco/models.py @@ -7,6 +7,31 @@ from pydantic import BaseModel, ConfigDict, Field from pydantic.alias_generators import to_camel +__all__ = [ + "Link", + "LinkList", + "LinkStats", + "ClickEvent", + "Folder", + "FolderList", + "BulkResult", + "BulkLinkResult", + "MeResponse", + "RoutingRule", + "OgMeta", + "GeoRestriction", + "QRSettings", + "TrustScoreResult", + "NamespaceInfo", + "NamespaceCheckResult", + "UtmTemplate", + "Webhook", + "SavedViewFilters", + "SavedView", + "CustomDomain", + "AffiliateProgram", +] + class _CamelModel(BaseModel): """Base model that accepts camelCase field names from the API.""" @@ -67,6 +92,7 @@ class LinkStats(_CamelModel): short_code: Optional[str] = None total_clicks: Optional[int] = None clicks: List[ClickEvent] = Field(default_factory=list) + aggregate_stats: Optional[Dict[str, Any]] = None # --------------------------------------------------------------------------- @@ -129,3 +155,180 @@ class MeResponse(_CamelModel): is_premium: Optional[bool] = None features: Optional[Dict[str, Any]] = None limits: Optional[Dict[str, Any]] = None + + +# --------------------------------------------------------------------------- +# Links — advanced field models +# --------------------------------------------------------------------------- + + +class RoutingRule(_CamelModel): + """A geo-routing rule for a link.""" + + country: Optional[str] = None + redirect_url: Optional[str] = None + + +class OgMeta(_CamelModel): + """Open Graph metadata override for a link.""" + + title: Optional[str] = None + description: Optional[str] = None + image: Optional[str] = None + + +class GeoRestriction(_CamelModel): + """Geo-restriction settings for a link.""" + + allowed_countries: Optional[List[str]] = None + blocked_countries: Optional[List[str]] = None + + +# --------------------------------------------------------------------------- +# QR Settings model +# --------------------------------------------------------------------------- + + +class QRSettings(_CamelModel): + """QR code settings for a link.""" + + size: Optional[int] = None + color: Optional[str] = None + bg_color: Optional[str] = None + error_correction: Optional[str] = None + margin: Optional[int] = None + logo_url: Optional[str] = None + + +# --------------------------------------------------------------------------- +# Trust Score model +# --------------------------------------------------------------------------- + + +class TrustScoreResult(_CamelModel): + """Result of a URL trust/safety scan.""" + + short: Optional[str] = None + long: Optional[str] = None + score: Optional[float] = None + status: Optional[str] = None + threats: Optional[List[str]] = None + scanned_at: Optional[str] = None + + +# --------------------------------------------------------------------------- +# Namespace models +# --------------------------------------------------------------------------- + + +class NamespaceInfo(_CamelModel): + """Namespace info for the authenticated user.""" + + has_access: Optional[bool] = None + namespace: Optional[str] = None + tier: Optional[str] = None + upgrade_required: Optional[bool] = None + + +class NamespaceCheckResult(_CamelModel): + """Result of checking namespace availability.""" + + namespace: Optional[str] = None + available: Optional[bool] = None + reason: Optional[str] = None + preview_url: Optional[str] = None + + +# --------------------------------------------------------------------------- +# UTM Template model +# --------------------------------------------------------------------------- + + +class UtmTemplate(_CamelModel): + """A saved UTM parameter template.""" + + id: Optional[str] = None + name: Optional[str] = None + source: Optional[str] = None + medium: Optional[str] = None + campaign: Optional[str] = None + term: Optional[str] = None + content: Optional[str] = None + + +# --------------------------------------------------------------------------- +# Webhook model +# --------------------------------------------------------------------------- + + +class Webhook(_CamelModel): + """A registered webhook endpoint.""" + + id: Optional[str] = None + url: Optional[str] = None + events: List[str] = Field(default_factory=list) + name: Optional[str] = None + enabled: Optional[bool] = None + created_at: Optional[str] = None + updated_at: Optional[str] = None + last_triggered: Optional[str] = None + failure_count: Optional[int] = None + + +# --------------------------------------------------------------------------- +# Saved View models +# --------------------------------------------------------------------------- + + +class SavedViewFilters(_CamelModel): + """Filter criteria for a saved view.""" + + folder_id: Optional[str] = None + tag: Optional[str] = None + status: Optional[str] = None + search: Optional[str] = None + date_range: Optional[str] = None + + +class SavedView(_CamelModel): + """A saved dashboard view with filter presets.""" + + id: Optional[str] = None + name: Optional[str] = None + filters: Optional[SavedViewFilters] = None + created_at: Optional[str] = None + updated_at: Optional[str] = None + + +# --------------------------------------------------------------------------- +# Custom Domain model +# --------------------------------------------------------------------------- + + +class CustomDomain(_CamelModel): + """A custom domain registered to the user's account.""" + + domain: Optional[str] = None + status: Optional[str] = None + verification_token: Optional[str] = None + is_default: Optional[bool] = None + link_count: Optional[int] = None + created_at: Optional[str] = None + + +# --------------------------------------------------------------------------- +# Affiliate Program model +# --------------------------------------------------------------------------- + + +class AffiliateProgram(_CamelModel): + """An affiliate program.""" + + id: Optional[str] = None + name: Optional[str] = None + description: Optional[str] = None + commission_type: Optional[str] = None + cpc_rate: Optional[float] = None + cpa_rate: Optional[float] = None + cookie_days: Optional[int] = None + status: Optional[str] = None diff --git a/awsysco/resources/affiliate.py b/awsysco/resources/affiliate.py new file mode 100644 index 0000000..d6549f9 --- /dev/null +++ b/awsysco/resources/affiliate.py @@ -0,0 +1,233 @@ +"""Affiliate resource — manage affiliate programs and partnerships.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .._http import HttpClient +from ..models import AffiliateProgram + + +class AffiliateResource: + """Interact with /api/affiliate endpoints.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def create_program( + self, + name: str, + commission_type: str, + **kwargs: Any, + ) -> AffiliateProgram: + """Create a new affiliate program. + + Args: + name: Display name for the program. + commission_type: Commission model — ``'cpc'``, ``'cpa_return'``, + or ``'both'``. + **kwargs: Optional fields: ``description``, ``cpc_rate`` (0.01–10), + ``cpa_rate`` (1–100), ``cookie_days``. + + Returns: + The created AffiliateProgram object. + """ + body: Dict[str, Any] = {"name": name, "commissionType": commission_type} + field_map = { + "description": "description", + "cpc_rate": "cpcRate", + "cpa_rate": "cpaRate", + "cookie_days": "cookieDays", + } + for k, v in kwargs.items(): + mapped = field_map.get(k, k) + body[mapped] = v + data = self._http.post("/api/affiliate/programs", json=body) + return AffiliateProgram.model_validate(data) + + def list_programs(self) -> List[AffiliateProgram]: + """List all affiliate programs owned by the authenticated user. + + Returns: + A list of AffiliateProgram objects. + """ + data = self._http.get("/api/affiliate/programs") + if isinstance(data, list): + return [AffiliateProgram.model_validate(item) for item in data] + items = data.get("programs", []) if isinstance(data, dict) else [] + return [AffiliateProgram.model_validate(item) for item in items] + + def get_program(self, program_id: str) -> AffiliateProgram: + """Get a single affiliate program by ID. + + Args: + program_id: The ID of the program. + + Returns: + The AffiliateProgram object. + """ + data = self._http.get(f"/api/affiliate/programs/{program_id}") + return AffiliateProgram.model_validate(data) + + def update_program(self, program_id: str, **kwargs: Any) -> AffiliateProgram: + """Update an affiliate program. + + Args: + program_id: The ID of the program to update. + **kwargs: Fields to update. Supported: ``name``, ``description``, + ``cpc_rate``, ``cpa_rate``, ``cookie_days``, ``status``. + + Returns: + The updated AffiliateProgram object. + """ + body: Dict[str, Any] = {} + field_map = { + "cpc_rate": "cpcRate", + "cpa_rate": "cpaRate", + "cookie_days": "cookieDays", + "commission_type": "commissionType", + } + for k, v in kwargs.items(): + mapped = field_map.get(k, k) + body[mapped] = v + data = self._http.patch(f"/api/affiliate/programs/{program_id}", json=body) + return AffiliateProgram.model_validate(data) + + def get_program_stats(self, program_id: str, *, period: str = "30d") -> dict: + """Get statistics for an affiliate program. + + Args: + program_id: The ID of the program. + period: Time period (default ``'30d'``). + + Returns: + API response dict with program statistics. + """ + return ( + self._http.get( + f"/api/affiliate/programs/{program_id}/stats", + params={"period": period}, + ) + or {} + ) + + def list_partners(self, program_id: str) -> List[dict]: + """List all partners for an affiliate program. + + Args: + program_id: The ID of the program. + + Returns: + A list of partner dicts. + """ + data = self._http.get(f"/api/affiliate/programs/{program_id}/partners") + if isinstance(data, list): + return data + return data.get("partners", []) if isinstance(data, dict) else [] + + def update_partner_status( + self, + program_id: str, + partner_id: str, + status: str, + ) -> dict: + """Update a partner's status within an affiliate program. + + Args: + program_id: The ID of the program. + partner_id: The ID of the partner. + status: New status string (e.g. ``'approved'``, ``'rejected'``). + + Returns: + API response dict. + """ + return ( + self._http.patch( + f"/api/affiliate/programs/{program_id}/partners/{partner_id}", + json={"status": status}, + ) + or {} + ) + + def discover(self, *, limit: int = 20) -> List[AffiliateProgram]: + """Discover public affiliate programs available to join. + + Args: + limit: Maximum number of programs to return (default 20). + + Returns: + A list of AffiliateProgram objects. + """ + data = self._http.get("/api/affiliate/discover", params={"limit": limit}) + if isinstance(data, list): + return [AffiliateProgram.model_validate(item) for item in data] + items = data.get("programs", []) if isinstance(data, dict) else [] + return [AffiliateProgram.model_validate(item) for item in items] + + def join(self, program_id: str, *, partner_code: Optional[str] = None) -> dict: + """Join a public affiliate program. + + Args: + program_id: The ID of the program to join. + partner_code: Optional referral/partner code. + + Returns: + API response dict with partnership details. + """ + body: Dict[str, Any] = {} + if partner_code is not None: + body["partnerCode"] = partner_code + return self._http.post(f"/api/affiliate/join/{program_id}", json=body) or {} + + def list_partnerships(self) -> List[dict]: + """List all affiliate partnerships the authenticated user has joined. + + Returns: + A list of partnership dicts. + """ + data = self._http.get("/api/affiliate/partnerships") + if isinstance(data, list): + return data + return data.get("partnerships", []) if isinstance(data, dict) else [] + + def get_partnership_stats( + self, + partnership_id: str, + *, + period: str = "30d", + ) -> dict: + """Get statistics for a specific partnership. + + Args: + partnership_id: The ID of the partnership. + period: Time period (default ``'30d'``). + + Returns: + API response dict with partnership statistics. + """ + return ( + self._http.get( + f"/api/affiliate/partnerships/{partnership_id}/stats", + params={"period": period}, + ) + or {} + ) + + def leave_program(self, partnership_id: str) -> dict: + """Leave an affiliate partnership. + + Args: + partnership_id: The ID of the partnership to leave. + + Returns: + API response dict. + """ + return self._http.delete(f"/api/affiliate/partnerships/{partnership_id}") or {} + + def get_limits(self) -> dict: + """Get the affiliate program limits for the authenticated user's tier. + + Returns: + API response dict with limit information. + """ + return self._http.get("/api/affiliate/limits") or {} diff --git a/awsysco/resources/agentlink.py b/awsysco/resources/agentlink.py new file mode 100644 index 0000000..63ff478 --- /dev/null +++ b/awsysco/resources/agentlink.py @@ -0,0 +1,61 @@ +"""AgentLink resource — AWSYS AgentLinks feature.""" + +from __future__ import annotations + +from .._http import HttpClient + + +class AgentlinkResource: + """Interact with /api/agentlink endpoints.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def subscribe(self, email: str) -> dict: + """Subscribe an email address to AgentLink updates. + + This is a public endpoint — no authentication required, though the + SDK will attach the Authorization header if an API key is configured. + + Args: + email: The email address to subscribe. + + Returns: + API response dict. + """ + return self._http.post("/api/agentlink/subscribe", json={"email": email}) or {} + + def get_link_stats(self, short_path: str, *, period_days: int = 7) -> dict: + """Get AgentLink click statistics for a specific link. + + Args: + short_path: The short code or slug identifying the link. + period_days: Number of days to include in the stats window (default 7). + + Returns: + API response dict with click statistics. + """ + return ( + self._http.get( + f"/api/agentlink/links/{short_path}/stats", + params={"period": period_days}, + ) + or {} + ) + + def get_account_stats(self, *, period_days: int = 7) -> dict: + """Get aggregated AgentLink statistics for the authenticated account. + + Args: + period_days: Number of days to include in the stats window (default 7). + + Returns: + API response dict with account-level statistics. + """ + return ( + self._http.get( + "/api/agentlink/account/stats", + params={"period": period_days}, + ) + or {} + ) diff --git a/awsysco/resources/analytics.py b/awsysco/resources/analytics.py index 78fff2b..91968a5 100644 --- a/awsysco/resources/analytics.py +++ b/awsysco/resources/analytics.py @@ -2,24 +2,56 @@ from __future__ import annotations +from typing import List, Optional + from .._http import HttpClient -from ..models import LinkStats +from ..models import ClickEvent, LinkStats class AnalyticsResource: - """Interact with /api/v1/links/:id/stats.""" + """Interact with /api/v1/links/:id/stats and analytics endpoints.""" def __init__(self, http: HttpClient) -> None: self._http = http - def get_stats(self, short_path: str) -> LinkStats: + def get_stats(self, short_path: str, *, period: Optional[str] = None) -> LinkStats: """Get click analytics for a link. Args: short_path: The short code or slug identifying the link. + period: Optional time period filter (e.g. ``'7d'``, ``'30d'``, + ``'all'``). Returns: A LinkStats object with total_clicks and per-click breakdown. """ - data = self._http.get(f"/api/v1/links/{short_path}/stats") + params = {} + if period is not None: + params["period"] = period + data = self._http.get( + f"/api/v1/links/{short_path}/stats", + params=params if params else None, + ) return LinkStats.model_validate(data) + + def get_recent_clicks(self, *, limit: Optional[int] = None) -> List[ClickEvent]: + """Get recent click events across all links for the authenticated user. + + Args: + limit: Maximum number of click events to return. + + Returns: + A list of ClickEvent objects. + """ + params = {} + if limit is not None: + params["limit"] = limit + data = self._http.get( + "/api/user/recent-clicks", + params=params if params else None, + ) + if isinstance(data, list): + return [ClickEvent.model_validate(item) for item in data] + # Some API shapes wrap in a key + items = data.get("clicks", data.get("recentClicks", [])) + return [ClickEvent.model_validate(item) for item in items] diff --git a/awsysco/resources/bulk.py b/awsysco/resources/bulk.py index d565c99..b3a2b8b 100644 --- a/awsysco/resources/bulk.py +++ b/awsysco/resources/bulk.py @@ -51,4 +51,10 @@ def create(self, urls: List[Dict[str, Any]]) -> BulkResult: payload.append(entry) data = self._http.post("/api/v1/bulk", json={"urls": payload}) + # Normalise: API sometimes wraps counts under a "summary" key + if isinstance(data, dict) and "summary" in data and "created" not in data: + summary = data["summary"] + data = dict(data) + data.setdefault("created", summary.get("created")) + data.setdefault("failed", summary.get("failed")) return BulkResult.model_validate(data) diff --git a/awsysco/resources/custom_domains.py b/awsysco/resources/custom_domains.py new file mode 100644 index 0000000..526c162 --- /dev/null +++ b/awsysco/resources/custom_domains.py @@ -0,0 +1,104 @@ +"""Custom Domains resource — manage custom short link domains.""" + +from __future__ import annotations + +from typing import Any, Dict, Optional + +from .._http import HttpClient +from ..models import CustomDomain + + +class CustomDomainsResource: + """Interact with /api/user/domains and /api/domains/check endpoints.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def list(self) -> dict: + """List all custom domains for the authenticated user. + + Returns: + API response dict containing domains. + """ + return self._http.get("/api/user/domains") or {} + + def add(self, domain: str) -> dict: + """Add a new custom domain. + + Args: + domain: The domain hostname to register (e.g. ``'links.example.com'``). + + Returns: + API response dict with verification token. + """ + return self._http.post("/api/user/domains", json={"domain": domain}) or {} + + def verify(self, domain: str) -> dict: + """Check DNS verification status for a domain. + + Args: + domain: The domain hostname to verify. + + Returns: + API response dict with verification status. + """ + return self._http.get(f"/api/user/domains/{domain}/verify") or {} + + def activate(self, domain: str) -> CustomDomain: + """Activate a verified domain. + + Args: + domain: The domain hostname to activate. + + Returns: + The activated CustomDomain object. + """ + data = self._http.post(f"/api/user/domains/{domain}/activate") + return CustomDomain.model_validate(data) + + def update( + self, + domain: str, + *, + is_default: Optional[bool] = None, + not_found_html: Optional[str] = None, + ) -> CustomDomain: + """Update custom domain settings. + + Args: + domain: The domain hostname to update. + is_default: Whether this domain should be the default. + not_found_html: Custom HTML for 404 pages on this domain. + + Returns: + The updated CustomDomain object. + """ + body: Dict[str, Any] = {} + if is_default is not None: + body["isDefault"] = is_default + if not_found_html is not None: + body["notFoundHtml"] = not_found_html + data = self._http.patch(f"/api/user/domains/{domain}", json=body) + return CustomDomain.model_validate(data) + + def remove(self, domain: str) -> dict: + """Remove a custom domain. + + Args: + domain: The domain hostname to remove. + + Returns: + API response dict. + """ + return self._http.delete(f"/api/user/domains/{domain}") or {} + + def check(self, hostname: str) -> dict: + """Check if a hostname is available for use as a custom domain. + + Args: + hostname: The hostname to check. + + Returns: + API response dict indicating availability. + """ + return self._http.get(f"/api/domains/check/{hostname}") or {} diff --git a/awsysco/resources/data_export.py b/awsysco/resources/data_export.py new file mode 100644 index 0000000..7bc60ff --- /dev/null +++ b/awsysco/resources/data_export.py @@ -0,0 +1,31 @@ +"""Data Export resource — CSV exports for links and link stats.""" + +from __future__ import annotations + +from .._http import HttpClient + + +class DataExportResource: + """Interact with /api/export endpoints that return CSV data.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def export_links(self) -> str: + """Export all links for the authenticated user as a CSV string. + + Returns: + A CSV-formatted string with all link data. + """ + return self._http.get_text("/api/export/links") + + def export_link_stats(self, short_path: str) -> str: + """Export click statistics for a specific link as a CSV string. + + Args: + short_path: The short code or slug identifying the link. + + Returns: + A CSV-formatted string with click statistics. + """ + return self._http.get_text(f"/api/export/stats/{short_path}") diff --git a/awsysco/resources/folders.py b/awsysco/resources/folders.py index 954007b..ce2ff26 100644 --- a/awsysco/resources/folders.py +++ b/awsysco/resources/folders.py @@ -40,6 +40,31 @@ def create(self, name: str, *, color: Optional[str] = None) -> Folder: data = self._http.post("/api/v1/folders", json=body) return Folder.model_validate(data) + def update( + self, + folder_id: str, + *, + name: Optional[str] = None, + color: Optional[str] = None, + ) -> Folder: + """Update a folder's name or color. + + Args: + folder_id: The ID of the folder to update. + name: New display name. + color: New hex color string. + + Returns: + The updated Folder object. + """ + body: Dict[str, Any] = {} + if name is not None: + body["name"] = name + if color is not None: + body["color"] = color + data = self._http.patch(f"/api/v1/folders/{folder_id}", json=body) + return Folder.model_validate(data) + def delete(self, folder_id: str) -> None: """Delete a folder. diff --git a/awsysco/resources/links.py b/awsysco/resources/links.py index 88d5c8e..344e66f 100644 --- a/awsysco/resources/links.py +++ b/awsysco/resources/links.py @@ -2,10 +2,10 @@ from __future__ import annotations -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from .._http import HttpClient -from ..models import Link, LinkList +from ..models import GeoRestriction, Link, LinkList, OgMeta, RoutingRule class LinksResource: @@ -21,6 +21,13 @@ def create( custom_slug: Optional[str] = None, expires_at: Optional[str] = None, max_clicks: Optional[int] = None, + routing_rules: Optional[List[Dict[str, str]]] = None, + og_meta: Optional[Dict[str, str]] = None, + geo_restriction: Optional[Dict[str, List[str]]] = None, + password: Optional[str] = None, + pass_ad_click_ids: Optional[bool] = None, + folder_id: Optional[str] = None, + tags: Optional[List[str]] = None, ) -> Link: """Create a new shortened link. @@ -29,6 +36,16 @@ def create( custom_slug: Optional custom short slug. expires_at: Optional expiry datetime (ISO 8601). max_clicks: Optional maximum click limit. + routing_rules: Optional list of geo-routing rules, each with + ``country`` and ``redirect_url`` keys. + og_meta: Optional Open Graph metadata dict with ``title``, + ``description``, and/or ``image`` keys. + geo_restriction: Optional dict with ``allowed_countries`` and/or + ``blocked_countries`` lists. + password: Optional password to protect the link. + pass_ad_click_ids: Whether to pass through ad click IDs (gclid etc). + folder_id: Optional folder ID to assign the link to. + tags: Optional list of tag strings. Returns: The created Link object. @@ -40,6 +57,20 @@ def create( body["expiresAt"] = expires_at if max_clicks is not None: body["maxClicks"] = max_clicks + if routing_rules is not None: + body["routingRules"] = routing_rules + if og_meta is not None: + body["ogMeta"] = og_meta + if geo_restriction is not None: + body["geoRestriction"] = geo_restriction + if password is not None: + body["password"] = password + if pass_ad_click_ids is not None: + body["passAdClickIds"] = pass_ad_click_ids + if folder_id is not None: + body["folderId"] = folder_id + if tags is not None: + body["tags"] = tags data = self._http.post("/api/v1/links", json=body) return Link.model_validate(data) @@ -73,24 +104,56 @@ def update( self, short_path: str, *, + url: Optional[str] = None, expires_at: Optional[str] = None, max_clicks: Optional[int] = None, + routing_rules: Optional[List[Dict[str, str]]] = None, + og_meta: Optional[Dict[str, str]] = None, + geo_restriction: Optional[Dict[str, List[str]]] = None, + password: Optional[str] = None, + pass_ad_click_ids: Optional[bool] = None, + folder_id: Optional[str] = None, + tags: Optional[List[str]] = None, ) -> Link: """Update a link's settings. Args: short_path: The short code or slug identifying the link. + url: New destination URL. expires_at: New expiry datetime (ISO 8601), or None to clear. max_clicks: New maximum click limit, or None to clear. + routing_rules: New list of geo-routing rules. + og_meta: New Open Graph metadata. + geo_restriction: New geo-restriction settings. + password: New password (or empty string to remove). + pass_ad_click_ids: Whether to pass through ad click IDs. + folder_id: New folder ID. + tags: New list of tags. Returns: The updated Link object. """ body: Dict[str, Any] = {} + if url is not None: + body["url"] = url if expires_at is not None: body["expiresAt"] = expires_at if max_clicks is not None: body["maxClicks"] = max_clicks + if routing_rules is not None: + body["routingRules"] = routing_rules + if og_meta is not None: + body["ogMeta"] = og_meta + if geo_restriction is not None: + body["geoRestriction"] = geo_restriction + if password is not None: + body["password"] = password + if pass_ad_click_ids is not None: + body["passAdClickIds"] = pass_ad_click_ids + if folder_id is not None: + body["folderId"] = folder_id + if tags is not None: + body["tags"] = tags data = self._http.patch(f"/api/v1/links/{short_path}", json=body) return Link.model_validate(data) diff --git a/awsysco/resources/namespace.py b/awsysco/resources/namespace.py new file mode 100644 index 0000000..a2c6b04 --- /dev/null +++ b/awsysco/resources/namespace.py @@ -0,0 +1,54 @@ +"""Namespace resource — branded namespace management.""" + +from __future__ import annotations + +from .._http import HttpClient +from ..models import NamespaceCheckResult, NamespaceInfo + + +class NamespaceResource: + """Interact with /api/user/namespace and /api/namespace/check endpoints.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def get(self) -> NamespaceInfo: + """Get the authenticated user's namespace info. + + Returns: + A NamespaceInfo object. + """ + data = self._http.get("/api/user/namespace") + return NamespaceInfo.model_validate(data) + + def check(self, namespace: str) -> NamespaceCheckResult: + """Check whether a namespace is available. + + Args: + namespace: The namespace string to check. + + Returns: + A NamespaceCheckResult indicating availability. + """ + data = self._http.get(f"/api/namespace/check/{namespace}") + return NamespaceCheckResult.model_validate(data) + + def claim(self, namespace: str) -> NamespaceInfo: + """Claim a namespace for the authenticated user. + + Args: + namespace: The namespace string to claim. + + Returns: + Updated NamespaceInfo after claiming. + """ + data = self._http.post("/api/user/namespace", json={"namespace": namespace}) + return NamespaceInfo.model_validate(data) + + def release(self) -> dict: + """Release the authenticated user's current namespace. + + Returns: + The API response dict. + """ + return self._http.delete("/api/user/namespace") or {} diff --git a/awsysco/resources/qr.py b/awsysco/resources/qr.py index afd4662..2c2823c 100644 --- a/awsysco/resources/qr.py +++ b/awsysco/resources/qr.py @@ -1,19 +1,19 @@ -"""QR code resource — URL construction for QR code images.""" +"""QR code resource — URL construction and settings CRUD for QR codes.""" from __future__ import annotations +from typing import Any, Dict from urllib.parse import urlencode from .._http import HttpClient +from ..models import QRSettings class QRResource: - """Build QR code image URLs for shortened links. - - No HTTP requests are made — this resource constructs the URL only. - """ + """Build QR code image URLs and manage QR code settings for shortened links.""" def __init__(self, http: HttpClient) -> None: + self._http = http self._base_url = http._base_url def get_url( @@ -26,6 +26,8 @@ def get_url( ) -> str: """Build the QR code image URL for a short code. + No HTTP request is made — this constructs the URL only. + Args: short_code: The short code/slug for the link. size: Image size in pixels (default 300). @@ -37,3 +39,30 @@ def get_url( """ params = urlencode({"size": size, "color": color, "bgColor": bg_color}) return f"{self._base_url}/api/qr/{short_code}?{params}" + + def get_settings(self, short_path: str) -> QRSettings: + """Get the QR code settings for a link. + + Args: + short_path: The short code or slug identifying the link. + + Returns: + A QRSettings object. + """ + data = self._http.get(f"/api/link/{short_path}/qr-settings") + return QRSettings.model_validate(data) + + def update_settings(self, short_path: str, settings: Dict[str, Any]) -> QRSettings: + """Update the QR code settings for a link. + + Args: + short_path: The short code or slug identifying the link. + settings: A dict of QR settings to apply. Valid keys include + ``size``, ``color``, ``bg_color``, ``error_correction``, + ``margin``, and ``logo_url``. + + Returns: + The updated QRSettings object. + """ + data = self._http.put(f"/api/link/{short_path}/qr-settings", json=settings) + return QRSettings.model_validate(data) diff --git a/awsysco/resources/saved_views.py b/awsysco/resources/saved_views.py new file mode 100644 index 0000000..c573604 --- /dev/null +++ b/awsysco/resources/saved_views.py @@ -0,0 +1,74 @@ +"""Saved Views resource — manage dashboard view presets.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .._http import HttpClient +from ..models import SavedView + + +class SavedViewsResource: + """Interact with /api/views endpoints.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def list(self) -> List[SavedView]: + """List all saved views for the authenticated user. + + Returns: + A list of SavedView objects. + """ + data = self._http.get("/api/views") + if isinstance(data, list): + return [SavedView.model_validate(item) for item in data] + items = data.get("views", []) if isinstance(data, dict) else [] + return [SavedView.model_validate(item) for item in items] + + def create(self, name: str, filters: Dict[str, Any]) -> SavedView: + """Create a new saved view. + + Args: + name: Display name for the view. + filters: Filter criteria dict. Supported keys: ``folderId``, + ``tag``, ``status``, ``search``, ``dateRange``. + + Returns: + The created SavedView object. + """ + data = self._http.post("/api/views", json={"name": name, "filters": filters}) + return SavedView.model_validate(data) + + def update( + self, + view_id: str, + *, + name: Optional[str] = None, + filters: Optional[Dict[str, Any]] = None, + ) -> SavedView: + """Update a saved view. + + Args: + view_id: The ID of the view to update. + name: New display name. + filters: New filter criteria dict. + + Returns: + The updated SavedView object. + """ + body: Dict[str, Any] = {} + if name is not None: + body["name"] = name + if filters is not None: + body["filters"] = filters + data = self._http.patch(f"/api/views/{view_id}", json=body) + return SavedView.model_validate(data) + + def delete(self, view_id: str) -> None: + """Delete a saved view. + + Args: + view_id: The ID of the view to delete. + """ + self._http.delete(f"/api/views/{view_id}") diff --git a/awsysco/resources/tags.py b/awsysco/resources/tags.py new file mode 100644 index 0000000..4383c07 --- /dev/null +++ b/awsysco/resources/tags.py @@ -0,0 +1,41 @@ +"""Tags resource — add and remove tags on links.""" + +from __future__ import annotations + +from urllib.parse import quote + +from .._http import HttpClient + + +class TagsResource: + """Interact with /api/link/:short/tags.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def add(self, short_path: str, tag: str) -> dict: + """Add a tag to a link. + + Args: + short_path: The short code or slug identifying the link. + tag: The tag string to add. + + Returns: + The API response dict. + """ + encoded = quote(short_path, safe="") + return self._http.post(f"/api/link/{encoded}/tags", json={"tag": tag}) or {} + + def remove(self, short_path: str, tag: str) -> dict: + """Remove a tag from a link. + + Args: + short_path: The short code or slug identifying the link. + tag: The tag string to remove. + + Returns: + The API response dict. + """ + encoded = quote(short_path, safe="") + encoded_tag = quote(tag, safe="") + return self._http.delete(f"/api/link/{encoded}/tags/{encoded_tag}") or {} diff --git a/awsysco/resources/trust_score.py b/awsysco/resources/trust_score.py new file mode 100644 index 0000000..4ad688a --- /dev/null +++ b/awsysco/resources/trust_score.py @@ -0,0 +1,28 @@ +"""Trust Score resource — URL safety scan.""" + +from __future__ import annotations + +from urllib.parse import quote + +from .._http import HttpClient +from ..models import TrustScoreResult + + +class TrustScoreResource: + """Interact with /api/link-scan/:short.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def scan(self, short_path: str) -> TrustScoreResult: + """Run a trust/safety scan on a shortened link. + + Args: + short_path: The short code or slug identifying the link. + + Returns: + A TrustScoreResult with score, status, and any detected threats. + """ + encoded = quote(short_path, safe="") + data = self._http.get(f"/api/link-scan/{encoded}") + return TrustScoreResult.model_validate(data) diff --git a/awsysco/resources/utm_templates.py b/awsysco/resources/utm_templates.py new file mode 100644 index 0000000..0c61a44 --- /dev/null +++ b/awsysco/resources/utm_templates.py @@ -0,0 +1,69 @@ +"""UTM Templates resource — manage saved UTM parameter templates.""" + +from __future__ import annotations + +from typing import Any, Dict, List + +from .._http import HttpClient +from ..models import UtmTemplate + + +class UtmTemplatesResource: + """Interact with UTM template endpoints.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def list(self) -> List[UtmTemplate]: + """List all saved UTM templates for the authenticated user. + + Returns: + A list of UtmTemplate objects. + """ + resp = self._http.get("/api/v1/me") + items = resp.get("utmTemplates", []) if isinstance(resp, dict) else [] + return [UtmTemplate.model_validate(item) for item in items] + + def create( + self, + name: str, + source: str, + medium: str, + campaign: str, + *, + term: str = "", + content: str = "", + ) -> dict: + """Create a new UTM template. + + Args: + name: Display name for the template. + source: UTM source value. + medium: UTM medium value. + campaign: UTM campaign value. + term: Optional UTM term value. + content: Optional UTM content value. + + Returns: + The API response dict. + """ + body: Dict[str, Any] = { + "name": name, + "source": source, + "medium": medium, + "campaign": campaign, + "term": term, + "content": content, + } + return self._http.post("/api/user/utm-templates", json=body) or {} + + def delete(self, template_id: str) -> dict: + """Delete a UTM template. + + Args: + template_id: The ID of the template to delete. + + Returns: + The API response dict. + """ + return self._http.delete(f"/api/user/utm-templates/{template_id}") or {} diff --git a/awsysco/resources/webhooks.py b/awsysco/resources/webhooks.py new file mode 100644 index 0000000..f513f7e --- /dev/null +++ b/awsysco/resources/webhooks.py @@ -0,0 +1,114 @@ +"""Webhooks resource — manage webhook endpoints and subscriptions.""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from .._http import HttpClient +from ..models import Webhook + + +class WebhooksResource: + """Interact with /api/webhooks endpoints.""" + + def __init__(self, http: HttpClient) -> None: + self._http = http + + def list_event_types(self) -> dict: + """List all available webhook event types. + + Returns: + API response dict containing supported event types. + """ + return self._http.get("/api/webhooks/event-types") or {} + + def list(self) -> dict: + """List all registered webhooks for the authenticated user. + + Returns: + API response dict containing webhooks. + """ + return self._http.get("/api/webhooks") or {} + + def create( + self, + url: str, + events: List[str], + *, + name: Optional[str] = None, + secret: Optional[str] = None, + ) -> Webhook: + """Register a new webhook. + + Args: + url: The HTTPS URL to deliver events to. + events: List of event type strings to subscribe to (e.g. + ``['link.created', 'link.click']``). + name: Optional human-readable name for this webhook. + secret: Optional signing secret for request verification. + + Returns: + The created Webhook object. + """ + body: Dict[str, Any] = {"url": url, "events": events} + if name is not None: + body["name"] = name + if secret is not None: + body["secret"] = secret + data = self._http.post("/api/webhooks", json=body) + return Webhook.model_validate(data) + + def update(self, webhook_id: str, **kwargs: Any) -> Webhook: + """Update a webhook's configuration. + + Args: + webhook_id: The ID of the webhook to update. + **kwargs: Fields to update. Supported keys: ``url``, ``events``, + ``name``, ``secret``, ``enabled``. + + Returns: + The updated Webhook object. + """ + # Map snake_case kwargs to camelCase + body: Dict[str, Any] = {} + key_map = { + "url": "url", + "events": "events", + "name": "name", + "secret": "secret", + "enabled": "enabled", + } + for k, v in kwargs.items(): + mapped = key_map.get(k, k) + body[mapped] = v + data = self._http.patch(f"/api/webhooks/{webhook_id}", json=body) + return Webhook.model_validate(data) + + def delete(self, webhook_id: str) -> dict: + """Delete a webhook. + + Args: + webhook_id: The ID of the webhook to delete. + + Returns: + The API response dict. + """ + return self._http.delete(f"/api/webhooks/{webhook_id}") or {} + + def test(self, webhook_id: str, event_type: str) -> dict: + """Send a test event to a webhook. + + Args: + webhook_id: The ID of the webhook to test. + event_type: The event type string to simulate. + + Returns: + The API response dict. + """ + return ( + self._http.post( + f"/api/webhooks/{webhook_id}/test", + json={"eventType": event_type}, + ) + or {} + ) diff --git a/examples/async_usage.py b/examples/async_usage.py new file mode 100644 index 0000000..af55925 --- /dev/null +++ b/examples/async_usage.py @@ -0,0 +1,83 @@ +"""Async usage examples for the AWSYS.CO Python SDK. + +Set AWSYS_API_KEY in your environment before running: + + export AWSYS_API_KEY=awsys_your_key_here + python examples/async_usage.py +""" + +import asyncio +import os + +from awsysco import AsyncClient, AwsysNotFoundError + +api_key = os.environ.get("AWSYS_API_KEY") +if not api_key: + raise SystemExit("Set AWSYS_API_KEY environment variable first.") + + +async def main() -> None: + async with AsyncClient(api_key=api_key) as client: + # ── Create a link ──────────────────────────────────────────────────── + print("Creating a short link (async)...") + link = await client.links.create( + "https://example.com/async/demo", + custom_slug=None, + max_clicks=1000, + ) + print(f" Short URL: {link.short_url}") + print(f" Short code: {link.short_code}") + + # ── List links ─────────────────────────────────────────────────────── + print("\nListing recent links (limit=5)...") + page = await client.links.list(limit=5) + print(f" Found {len(page.links)} links (total: {page.total})") + + # ── Analytics ──────────────────────────────────────────────────────── + if link.short_code: + print(f"\nFetching stats for {link.short_code!r}...") + stats = await client.analytics.get_stats(link.short_code, period="7d") + print(f" Total clicks: {stats.total_clicks}") + + # ── QR code URL ────────────────────────────────────────────────────── + if link.short_code: + qr_url = client.qr.get_url(link.short_code, size=400, color="1A1A2E") + print(f"\nQR code URL: {qr_url}") + + # ── Webhooks ───────────────────────────────────────────────────────── + print("\nListing available webhook event types...") + event_types = await client.webhooks.list_event_types() + print(f" Event types: {event_types}") + + # ── Namespace ──────────────────────────────────────────────────────── + print("\nChecking namespace info...") + ns = await client.namespace.get() + print(f" Has access: {ns.has_access}") + print(f" Namespace: {ns.namespace}") + + # ── Me ─────────────────────────────────────────────────────────────── + print("\nFetching user info...") + me = await client.me.get() + print(f" Email: {me.email}") + print(f" Tier: {me.subscription_tier}") + + # ── UTM templates ──────────────────────────────────────────────────── + print("\nListing UTM templates...") + templates = await client.utm_templates.list() + print(f" Found {len(templates)} templates") + + # ── Saved views ────────────────────────────────────────────────────── + print("\nListing saved views...") + views = await client.saved_views.list() + print(f" Found {len(views)} saved views") + + # ── Cleanup ────────────────────────────────────────────────────────── + print("\nCleaning up...") + if link.short_code: + await client.links.delete(link.short_code) + print(f" Deleted link {link.short_code!r}") + + print("Done.") + + +asyncio.run(main()) diff --git a/examples/integration_test.py b/examples/integration_test.py new file mode 100644 index 0000000..945a3da --- /dev/null +++ b/examples/integration_test.py @@ -0,0 +1,398 @@ +"""Manual integration test script for the AWSYS.CO Python SDK. + +Tests each major operation with PASS/FAIL output. NOT run by pytest — +run manually after setting AWSYS_API_KEY: + + export AWSYS_API_KEY=awsys_your_key_here + python examples/integration_test.py +""" + +from __future__ import annotations + +import os +import time + +from awsysco import Client, AwsysError, AwsysNotFoundError +from awsysco.models import ( + AffiliateProgram, + BulkResult, + ClickEvent, + CustomDomain, + Folder, + FolderList, + Link, + LinkList, + LinkStats, + MeResponse, + NamespaceCheckResult, + NamespaceInfo, + SavedView, + TrustScoreResult, + UtmTemplate, + Webhook, +) + +api_key = os.environ.get("AWSYS_API_KEY") +if not api_key: + raise SystemExit("Set AWSYS_API_KEY environment variable first.") + +base_url = os.environ.get("AWSYS_BASE_URL", "https://awsys.co") +client = Client(api_key=api_key, base_url=base_url) + +_passed = 0 +_failed = 0 +_skipped = 0 + + +def _ts() -> str: + return str(int(time.time() * 1000)) + + +def check(name: str, condition: bool, detail: str = "") -> None: + global _passed, _failed + if condition: + _passed += 1 + print(f" PASS {name}") + else: + _failed += 1 + print(f" FAIL {name}" + (f" — {detail}" if detail else "")) + + +def skip(name: str, reason: str) -> None: + global _skipped + _skipped += 1 + print(f" SKIP {name} — {reason}") + + +def section(title: str) -> None: + print(f"\n{'─' * 60}") + print(f" {title}") + print(f"{'─' * 60}") + + +# ── Me ──────────────────────────────────────────────────────────────────────── +section("Me") +try: + me = client.me.get() + check("get() returns MeResponse", isinstance(me, MeResponse)) + check("email is present", me.email is not None) + check("subscription_tier is present", me.subscription_tier is not None) +except AwsysError as e: + check("me.get()", False, str(e)) + +# ── Links ───────────────────────────────────────────────────────────────────── +section("Links") +created_link: Link | None = None +try: + created_link = client.links.create(f"https://example.com/sdk-test-{_ts()}") + check("create() returns Link", isinstance(created_link, Link)) + check("short_url is present", created_link.short_url is not None) + check("short_code is present", created_link.short_code is not None) +except AwsysError as e: + skip("create()", str(e)) + +try: + page = client.links.list(limit=5) + check("list() returns LinkList", isinstance(page, LinkList)) + check("links is a list", isinstance(page.links, list)) +except AwsysError as e: + check("list()", False, str(e)) + +if created_link and created_link.short_code: + try: + fetched = client.links.get(created_link.short_code) + check("get() returns Link", isinstance(fetched, Link)) + check("get() returns correct code", fetched.short_code == created_link.short_code) + except AwsysError as e: + check("get()", False, str(e)) + + try: + updated = client.links.update(created_link.short_code, max_clicks=999) + check("update() returns Link", isinstance(updated, Link)) + check("update() applies max_clicks", updated.max_clicks == 999) + except AwsysError as e: + check("update()", False, str(e)) + +# ── Analytics ───────────────────────────────────────────────────────────────── +section("Analytics") +if created_link and created_link.short_code: + try: + stats = client.analytics.get_stats(created_link.short_code) + check("get_stats() returns LinkStats", isinstance(stats, LinkStats)) + check("total_clicks is int", isinstance(stats.total_clicks, int)) + except AwsysError as e: + check("get_stats()", False, str(e)) +else: + skip("get_stats()", "No link available") + +try: + clicks = client.analytics.get_recent_clicks(limit=5) + check("get_recent_clicks() returns list", isinstance(clicks, list)) + check("items are ClickEvent", all(isinstance(c, ClickEvent) for c in clicks)) +except AwsysError as e: + check("get_recent_clicks()", False, str(e)) + +# ── QR ──────────────────────────────────────────────────────────────────────── +section("QR") +if created_link and created_link.short_code: + qr_url = client.qr.get_url(created_link.short_code, size=300) + check("get_url() returns string", isinstance(qr_url, str)) + check("get_url() contains short code", created_link.short_code in qr_url) +else: + skip("get_url()", "No link available") + +# ── Folders ─────────────────────────────────────────────────────────────────── +section("Folders") +created_folder: Folder | None = None +try: + created_folder = client.folders.create(f"SDK Test {_ts()}", color="#3498DB") + check("create() returns Folder", isinstance(created_folder, Folder)) + check("id is present", created_folder.id is not None) +except AwsysError as e: + check("create()", False, str(e)) + +try: + folder_list = client.folders.list() + check("list() returns FolderList", isinstance(folder_list, FolderList)) +except AwsysError as e: + check("list()", False, str(e)) + +if created_folder and created_folder.id: + try: + updated_folder = client.folders.update(created_folder.id, name=f"Renamed {_ts()}") + check("update() returns Folder", isinstance(updated_folder, Folder)) + except AwsysError as e: + check("update()", False, str(e)) + +# ── Bulk ────────────────────────────────────────────────────────────────────── +section("Bulk") +try: + ts = _ts() + result = client.bulk.create([ + {"url": f"https://example.com/bulk-{ts}-0"}, + {"url": f"https://example.com/bulk-{ts}-1"}, + ]) + check("create() returns BulkResult", isinstance(result, BulkResult)) + if result.results and all(r.success for r in result.results): + check("all items succeeded", True) + else: + skip("all items succeeded", "Some items failed (account restriction?)") +except AwsysError as e: + check("bulk.create()", False, str(e)) + +# ── Webhooks ────────────────────────────────────────────────────────────────── +section("Webhooks") +try: + event_types = client.webhooks.list_event_types() + check("list_event_types() returns dict", isinstance(event_types, dict)) +except AwsysError as e: + check("list_event_types()", False, str(e)) + +try: + webhooks = client.webhooks.list() + check("list() returns dict", isinstance(webhooks, dict)) +except AwsysError as e: + check("list()", False, str(e)) + +created_webhook: Webhook | None = None +try: + created_webhook = client.webhooks.create( + url="https://example.com/webhook", + events=["link.created"], + name=f"SDK Test {_ts()}", + ) + check("create() returns Webhook", isinstance(created_webhook, Webhook)) + check("webhook id is present", created_webhook.id is not None) +except AwsysError as e: + skip("create()", str(e)) + +if created_webhook and created_webhook.id: + try: + updated_wh = client.webhooks.update(created_webhook.id, enabled=False) + check("update() returns Webhook", isinstance(updated_wh, Webhook)) + except AwsysError as e: + check("update()", False, str(e)) + + try: + client.webhooks.delete(created_webhook.id) + check("delete() succeeds", True) + except AwsysError as e: + check("delete()", False, str(e)) + +# ── Namespace ───────────────────────────────────────────────────────────────── +section("Namespace") +try: + ns = client.namespace.get() + check("get() returns NamespaceInfo", isinstance(ns, NamespaceInfo)) +except AwsysError as e: + check("get()", False, str(e)) + +try: + result_check = client.namespace.check(f"sdktest{_ts()[-6:]}") + check("check() returns NamespaceCheckResult", isinstance(result_check, NamespaceCheckResult)) + check("available field is bool", isinstance(result_check.available, bool)) +except AwsysError as e: + check("check()", False, str(e)) + +# ── UTM Templates ───────────────────────────────────────────────────────────── +section("UTM Templates") +try: + templates = client.utm_templates.list() + check("list() returns list", isinstance(templates, list)) + check("items are UtmTemplate", all(isinstance(t, UtmTemplate) for t in templates)) +except AwsysError as e: + check("list()", False, str(e)) + +try: + resp = client.utm_templates.create( + name=f"SDK Test {_ts()}", + source="sdk", + medium="test", + campaign="integration", + ) + check("create() returns dict", isinstance(resp, dict)) +except AwsysError as e: + check("create()", False, str(e)) + +# ── Saved Views ─────────────────────────────────────────────────────────────── +section("Saved Views") +try: + views = client.saved_views.list() + check("list() returns list", isinstance(views, list)) + check("items are SavedView", all(isinstance(v, SavedView) for v in views)) +except AwsysError as e: + check("list()", False, str(e)) + +created_view: SavedView | None = None +try: + created_view = client.saved_views.create( + name=f"SDK Test {_ts()}", + filters={"status": "active"}, + ) + check("create() returns SavedView", isinstance(created_view, SavedView)) + check("view id is present", created_view.id is not None) +except AwsysError as e: + skip("create()", str(e)) + +if created_view and created_view.id: + try: + client.saved_views.delete(created_view.id) + check("delete() succeeds", True) + except AwsysError as e: + check("delete()", False, str(e)) + +# ── Custom Domains ──────────────────────────────────────────────────────────── +section("Custom Domains") +try: + domains = client.custom_domains.list() + check("list() returns dict", isinstance(domains, dict)) +except AwsysError as e: + check("list()", False, str(e)) + +try: + avail = client.custom_domains.check("sdk-test-never-exists.example.com") + check("check() returns dict", isinstance(avail, dict)) +except AwsysError as e: + check("check()", False, str(e)) + +# ── Tags ────────────────────────────────────────────────────────────────────── +section("Tags") +if created_link and created_link.short_code: + try: + resp = client.tags.add(created_link.short_code, "sdk-test") + check("add() returns dict", isinstance(resp, dict)) + except AwsysError as e: + check("add()", False, str(e)) + + try: + resp = client.tags.remove(created_link.short_code, "sdk-test") + check("remove() returns dict", isinstance(resp, dict)) + except AwsysError as e: + check("remove()", False, str(e)) +else: + skip("tags", "No link available") + +# ── Trust Score ─────────────────────────────────────────────────────────────── +section("Trust Score") +if created_link and created_link.short_code: + try: + result_ts = client.trust_score.scan(created_link.short_code) + check("scan() returns TrustScoreResult", isinstance(result_ts, TrustScoreResult)) + except AwsysError as e: + skip("scan()", str(e)) +else: + skip("scan()", "No link available") + +# ── Agentlink ───────────────────────────────────────────────────────────────── +section("Agentlink") +try: + resp = client.agentlink.subscribe("sdk-test@example.com") + check("subscribe() returns dict", isinstance(resp, dict)) +except AwsysError as e: + skip("subscribe()", str(e)) + +try: + stats = client.agentlink.get_account_stats(period_days=7) + check("get_account_stats() returns dict", isinstance(stats, dict)) +except AwsysError as e: + skip("get_account_stats()", str(e)) + +# ── Affiliate ───────────────────────────────────────────────────────────────── +section("Affiliate") +try: + programs = client.affiliate.list_programs() + check("list_programs() returns list", isinstance(programs, list)) +except AwsysError as e: + check("list_programs()", False, str(e)) + +try: + discover = client.affiliate.discover(limit=5) + check("discover() returns list", isinstance(discover, list)) +except AwsysError as e: + check("discover()", False, str(e)) + +try: + limits = client.affiliate.get_limits() + check("get_limits() returns dict", isinstance(limits, dict)) +except AwsysError as e: + check("get_limits()", False, str(e)) + +try: + partnerships = client.affiliate.list_partnerships() + check("list_partnerships() returns list", isinstance(partnerships, list)) +except AwsysError as e: + check("list_partnerships()", False, str(e)) + +# ── Data Export ─────────────────────────────────────────────────────────────── +section("Data Export") +try: + csv_data = client.data_export.export_links() + check("export_links() returns string", isinstance(csv_data, str)) + check("export_links() is non-empty", len(csv_data) > 0) +except AwsysError as e: + check("export_links()", False, str(e)) + +# ── Cleanup ─────────────────────────────────────────────────────────────────── +section("Cleanup") +if created_link and created_link.short_code: + try: + client.links.delete(created_link.short_code) + check("delete link", True) + except AwsysError as e: + check("delete link", False, str(e)) + +if created_folder and created_folder.id: + try: + client.folders.delete(created_folder.id) + check("delete folder", True) + except AwsysError as e: + check("delete folder", False, str(e)) + +# ── Summary ─────────────────────────────────────────────────────────────────── +total = _passed + _failed + _skipped +print(f"\n{'=' * 60}") +print(f" Results: {_passed} passed / {_failed} failed / {_skipped} skipped ({total} total)") +print(f"{'=' * 60}\n") + +if _failed > 0: + raise SystemExit(1) diff --git a/pyproject.toml b/pyproject.toml index 4d7faec..0a439bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "awsysco" -version = "0.1.0" +version = "1.0.0" description = "Official Python SDK for the AWSYS.CO URL Shortener API" readme = "README.md" requires-python = ">=3.9" diff --git a/tests/conftest.py b/tests/conftest.py index 1cc12df..f564b08 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,10 +6,17 @@ from dotenv import load_dotenv from awsysco import Client +from awsysco.exceptions import AwsysForbiddenError # Load staging credentials from .env.test (gitignored) load_dotenv(".env.test") +# Messages that indicate a staging account restriction rather than a code bug +_SKIP_MESSAGES = ( + "email verification required", + "email not verified", +) + @pytest.fixture(scope="session") def client() -> Client: @@ -20,3 +27,23 @@ def client() -> Client: base_url = os.environ.get("AWSYS_BASE_URL", "https://staging.awsys.co") return Client(api_key=api_key, base_url=base_url) + + +@pytest.fixture(autouse=True) +def skip_on_account_restriction(request): + """Skip integration tests that fail due to staging account restrictions + (e.g. email verification required) rather than code bugs. + """ + yield + # Nothing to do post-yield — we catch before yield via try/except below + + +def pytest_runtest_call(item): + """Hook: convert AwsysForbiddenError 'email verification required' into a skip.""" + try: + item.runtest() + except AwsysForbiddenError as exc: + msg = str(exc).lower() + if any(phrase in msg for phrase in _SKIP_MESSAGES): + pytest.skip(f"Staging account restriction: {exc}") + raise diff --git a/tests/test_affiliate.py b/tests/test_affiliate.py new file mode 100644 index 0000000..02b6b26 --- /dev/null +++ b/tests/test_affiliate.py @@ -0,0 +1,162 @@ +"""Unit tests for the Affiliate resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.models import AffiliateProgram +from awsysco.resources.affiliate import AffiliateResource + + +_PROGRAM_DATA = { + "id": "prog1", + "name": "My Affiliate Program", + "description": "Great program", + "commissionType": "cpc", + "cpcRate": 0.5, + "cpaRate": None, + "cookieDays": 30, + "status": "active", +} + + +def _make_resource(): + http = MagicMock() + http.post.return_value = _PROGRAM_DATA + http.get.return_value = [_PROGRAM_DATA] + http.patch.return_value = _PROGRAM_DATA + http.delete.return_value = None + return AffiliateResource(http) + + +class TestAffiliatePrograms: + def test_create_program_calls_endpoint(self): + resource = _make_resource() + resource.create_program("My Program", "cpc", cpc_rate=0.5) + resource._http.post.assert_called_once() + body = resource._http.post.call_args[1]["json"] + assert body["name"] == "My Program" + assert body["commissionType"] == "cpc" + assert body["cpcRate"] == 0.5 + + def test_create_program_returns_affiliate_program(self): + resource = _make_resource() + result = resource.create_program("My Program", "cpc") + assert isinstance(result, AffiliateProgram) + assert result.id == "prog1" + + def test_list_programs_calls_endpoint(self): + resource = _make_resource() + resource.list_programs() + resource._http.get.assert_called_once_with("/api/affiliate/programs") + + def test_list_programs_returns_list(self): + resource = _make_resource() + result = resource.list_programs() + assert isinstance(result, list) + assert all(isinstance(p, AffiliateProgram) for p in result) + + def test_list_programs_handles_dict_wrapper(self): + resource = _make_resource() + resource._http.get.return_value = {"programs": [_PROGRAM_DATA]} + result = resource.list_programs() + assert len(result) == 1 + + def test_get_program_calls_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = _PROGRAM_DATA + resource.get_program("prog1") + resource._http.get.assert_called_once_with("/api/affiliate/programs/prog1") + + def test_get_program_returns_affiliate_program(self): + resource = _make_resource() + resource._http.get.return_value = _PROGRAM_DATA + result = resource.get_program("prog1") + assert isinstance(result, AffiliateProgram) + + def test_update_program_maps_fields(self): + resource = _make_resource() + resource.update_program("prog1", cpc_rate=1.0, cookie_days=60) + body = resource._http.patch.call_args[1]["json"] + assert body["cpcRate"] == 1.0 + assert body["cookieDays"] == 60 + + def test_get_program_stats_calls_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = {"clicks": 100} + resource.get_program_stats("prog1") + resource._http.get.assert_called_once_with( + "/api/affiliate/programs/prog1/stats", params={"period": "30d"} + ) + + def test_list_partners_calls_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = [] + resource.list_partners("prog1") + resource._http.get.assert_called_once_with( + "/api/affiliate/programs/prog1/partners" + ) + + def test_update_partner_status_calls_patch(self): + resource = _make_resource() + resource.update_partner_status("prog1", "partner1", "approved") + resource._http.patch.assert_called_once_with( + "/api/affiliate/programs/prog1/partners/partner1", + json={"status": "approved"}, + ) + + def test_discover_calls_endpoint(self): + resource = _make_resource() + resource.discover() + resource._http.get.assert_called_once_with( + "/api/affiliate/discover", params={"limit": 20} + ) + + def test_discover_custom_limit(self): + resource = _make_resource() + resource.discover(limit=5) + resource._http.get.assert_called_once_with( + "/api/affiliate/discover", params={"limit": 5} + ) + + def test_join_calls_endpoint(self): + resource = _make_resource() + resource._http.post.return_value = {"joined": True} + resource.join("prog1") + resource._http.post.assert_called_once_with( + "/api/affiliate/join/prog1", json={} + ) + + def test_join_with_partner_code(self): + resource = _make_resource() + resource._http.post.return_value = {"joined": True} + resource.join("prog1", partner_code="CODE123") + body = resource._http.post.call_args[1]["json"] + assert body["partnerCode"] == "CODE123" + + def test_list_partnerships_calls_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = [] + resource.list_partnerships() + resource._http.get.assert_called_once_with("/api/affiliate/partnerships") + + def test_get_partnership_stats_calls_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = {} + resource.get_partnership_stats("p1") + resource._http.get.assert_called_once_with( + "/api/affiliate/partnerships/p1/stats", params={"period": "30d"} + ) + + def test_leave_program_calls_delete(self): + resource = _make_resource() + resource.leave_program("p1") + resource._http.delete.assert_called_once_with("/api/affiliate/partnerships/p1") + + def test_get_limits_calls_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = {"maxPrograms": 3} + resource.get_limits() + resource._http.get.assert_called_once_with("/api/affiliate/limits") diff --git a/tests/test_agentlink.py b/tests/test_agentlink.py new file mode 100644 index 0000000..3c5816a --- /dev/null +++ b/tests/test_agentlink.py @@ -0,0 +1,58 @@ +"""Unit tests for the AgentLink resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.resources.agentlink import AgentlinkResource + + +def _make_resource(): + http = MagicMock() + http.post.return_value = {"subscribed": True} + http.get.return_value = {"clicks": 42} + return AgentlinkResource(http) + + +class TestAgentlink: + def test_subscribe_calls_correct_endpoint(self): + resource = _make_resource() + resource.subscribe("user@example.com") + resource._http.post.assert_called_once_with( + "/api/agentlink/subscribe", json={"email": "user@example.com"} + ) + + def test_subscribe_returns_dict(self): + resource = _make_resource() + result = resource.subscribe("user@example.com") + assert isinstance(result, dict) + + def test_get_link_stats_calls_correct_endpoint(self): + resource = _make_resource() + resource.get_link_stats("abc123") + resource._http.get.assert_called_once_with( + "/api/agentlink/links/abc123/stats", params={"period": 7} + ) + + def test_get_link_stats_custom_period(self): + resource = _make_resource() + resource.get_link_stats("abc123", period_days=30) + resource._http.get.assert_called_once_with( + "/api/agentlink/links/abc123/stats", params={"period": 30} + ) + + def test_get_account_stats_calls_correct_endpoint(self): + resource = _make_resource() + resource.get_account_stats() + resource._http.get.assert_called_once_with( + "/api/agentlink/account/stats", params={"period": 7} + ) + + def test_get_account_stats_custom_period(self): + resource = _make_resource() + resource.get_account_stats(period_days=14) + resource._http.get.assert_called_once_with( + "/api/agentlink/account/stats", params={"period": 14} + ) diff --git a/tests/test_analytics.py b/tests/test_analytics.py index 1d5a989..0afde12 100644 --- a/tests/test_analytics.py +++ b/tests/test_analytics.py @@ -1,9 +1,100 @@ -"""Integration tests for the Analytics resource.""" +"""Tests for the Analytics resource.""" + +from __future__ import annotations import time +from unittest.mock import MagicMock from awsysco import Client -from awsysco.models import LinkStats +from awsysco.models import ClickEvent, LinkStats +from awsysco.resources.analytics import AnalyticsResource + + +# --------------------------------------------------------------------------- +# Unit tests — no network required +# --------------------------------------------------------------------------- + +_STATS_DATA = { + "shortCode": "abc", + "totalClicks": 42, + "clicks": [ + {"timestamp": "2026-01-01T00:00:00Z", "country": "US", "device": "desktop"}, + ], + "aggregateStats": {"byCountry": {"US": 42}}, +} + + +def _make_resource(): + http = MagicMock() + http.get.return_value = _STATS_DATA + return AnalyticsResource(http) + + +class TestAnalyticsUnit: + def test_get_stats_calls_correct_endpoint(self): + resource = _make_resource() + resource.get_stats("abc") + resource._http.get.assert_called_once_with( + "/api/v1/links/abc/stats", params=None + ) + + def test_get_stats_with_period(self): + resource = _make_resource() + resource.get_stats("abc", period="7d") + resource._http.get.assert_called_once_with( + "/api/v1/links/abc/stats", params={"period": "7d"} + ) + + def test_get_stats_returns_link_stats(self): + resource = _make_resource() + result = resource.get_stats("abc") + assert isinstance(result, LinkStats) + assert result.total_clicks == 42 + + def test_get_stats_populates_aggregate_stats(self): + resource = _make_resource() + result = resource.get_stats("abc") + assert result.aggregate_stats is not None + assert "byCountry" in result.aggregate_stats + + def test_get_recent_clicks_calls_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = [] + resource.get_recent_clicks() + resource._http.get.assert_called_once_with( + "/api/user/recent-clicks", params=None + ) + + def test_get_recent_clicks_with_limit(self): + resource = _make_resource() + resource._http.get.return_value = [] + resource.get_recent_clicks(limit=10) + resource._http.get.assert_called_once_with( + "/api/user/recent-clicks", params={"limit": 10} + ) + + def test_get_recent_clicks_returns_list(self): + resource = _make_resource() + resource._http.get.return_value = [ + {"timestamp": "2026-01-01T00:00:00Z", "country": "US"} + ] + result = resource.get_recent_clicks() + assert isinstance(result, list) + assert all(isinstance(c, ClickEvent) for c in result) + + def test_get_recent_clicks_handles_wrapped_response(self): + resource = _make_resource() + resource._http.get.return_value = { + "clicks": [{"timestamp": "2026-01-01T00:00:00Z"}] + } + result = resource.get_recent_clicks() + assert len(result) == 1 + assert isinstance(result[0], ClickEvent) + + +# --------------------------------------------------------------------------- +# Integration tests — require AWSYS_API_KEY +# --------------------------------------------------------------------------- def _unique_url() -> str: diff --git a/tests/test_async_client.py b/tests/test_async_client.py new file mode 100644 index 0000000..d96edbe --- /dev/null +++ b/tests/test_async_client.py @@ -0,0 +1,96 @@ +"""Unit tests for AsyncClient construction and resource wiring.""" + +from __future__ import annotations + +import pytest + +from awsysco import AsyncClient + + +class TestAsyncClientConstruction: + def test_instantiates_without_error(self): + client = AsyncClient(api_key="awsys_test") + assert client is not None + + def test_has_links_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "links") + + def test_has_analytics_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "analytics") + + def test_has_qr_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "qr") + + def test_has_folders_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "folders") + + def test_has_bulk_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "bulk") + + def test_has_me_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "me") + + def test_has_tags_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "tags") + + def test_has_trust_score_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "trust_score") + + def test_has_data_export_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "data_export") + + def test_has_namespace_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "namespace") + + def test_has_utm_templates_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "utm_templates") + + def test_has_webhooks_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "webhooks") + + def test_has_saved_views_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "saved_views") + + def test_has_custom_domains_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "custom_domains") + + def test_has_agentlink_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "agentlink") + + def test_has_affiliate_resource(self): + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "affiliate") + + def test_qr_get_url_is_synchronous(self): + """QR URL construction is sync even in AsyncClient.""" + client = AsyncClient(api_key="awsys_test", base_url="https://awsys.co") + url = client.qr.get_url("abc123") + assert "abc123" in url + assert url.startswith("https://awsys.co") + + def test_custom_base_url(self): + client = AsyncClient(api_key="awsys_test", base_url="https://staging.awsys.co") + url = client.qr.get_url("test") + assert "staging.awsys.co" in url + + def test_is_async_context_manager(self): + """AsyncClient exposes __aenter__ and __aexit__.""" + client = AsyncClient(api_key="awsys_test") + assert hasattr(client, "__aenter__") + assert hasattr(client, "__aexit__") + assert hasattr(client, "aclose") diff --git a/tests/test_bulk.py b/tests/test_bulk.py index 07fe904..df1a017 100644 --- a/tests/test_bulk.py +++ b/tests/test_bulk.py @@ -2,6 +2,8 @@ import time +import pytest + from awsysco import Client from awsysco.models import BulkResult @@ -11,6 +13,13 @@ def _unique_urls(count: int = 3) -> list: return [{"url": f"https://example.com/sdk-bulk-{ts}-{i}"} for i in range(count)] +def _assert_bulk_not_restricted(result: BulkResult) -> None: + """Skip if all results failed due to account restrictions (e.g. unverified email).""" + if result.results and all(not r.success for r in result.results): + errors = {r.error for r in result.results if r.error} + pytest.skip(f"Staging account restriction — all bulk items failed: {errors}") + + class TestBulk: def test_bulk_create_returns_result(self, client: Client) -> None: urls = _unique_urls(3) @@ -20,11 +29,13 @@ def test_bulk_create_returns_result(self, client: Client) -> None: def test_bulk_create_count(self, client: Client) -> None: urls = _unique_urls(3) result = client.bulk.create(urls) + _assert_bulk_not_restricted(result) assert result.created == 3 def test_bulk_results_have_short_url(self, client: Client) -> None: urls = _unique_urls(3) result = client.bulk.create(urls) + _assert_bulk_not_restricted(result) assert len(result.results) == 3 for item in result.results: assert item.success is True diff --git a/tests/test_custom_domains.py b/tests/test_custom_domains.py new file mode 100644 index 0000000..261221b --- /dev/null +++ b/tests/test_custom_domains.py @@ -0,0 +1,96 @@ +"""Unit tests for the Custom Domains resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.models import CustomDomain +from awsysco.resources.custom_domains import CustomDomainsResource + + +_DOMAIN_DATA = { + "domain": "links.example.com", + "status": "active", + "verificationToken": "abc123", + "isDefault": False, + "linkCount": 0, + "createdAt": "2026-01-01T00:00:00Z", +} + + +def _make_resource(): + http = MagicMock() + http.get.return_value = {"domains": [_DOMAIN_DATA]} + http.post.return_value = _DOMAIN_DATA + http.patch.return_value = _DOMAIN_DATA + http.delete.return_value = None + return CustomDomainsResource(http) + + +class TestCustomDomains: + def test_list_calls_endpoint(self): + resource = _make_resource() + resource.list() + resource._http.get.assert_called_once_with("/api/user/domains") + + def test_add_calls_endpoint(self): + resource = _make_resource() + resource.add("links.example.com") + resource._http.post.assert_called_once_with( + "/api/user/domains", json={"domain": "links.example.com"} + ) + + def test_verify_calls_correct_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = {"verified": True} + resource.verify("links.example.com") + resource._http.get.assert_called_once_with( + "/api/user/domains/links.example.com/verify" + ) + + def test_activate_calls_correct_endpoint(self): + resource = _make_resource() + resource.activate("links.example.com") + resource._http.post.assert_called_once_with( + "/api/user/domains/links.example.com/activate" + ) + + def test_activate_returns_custom_domain(self): + resource = _make_resource() + result = resource.activate("links.example.com") + assert isinstance(result, CustomDomain) + assert result.domain == "links.example.com" + + def test_update_sends_is_default(self): + resource = _make_resource() + resource.update("links.example.com", is_default=True) + body = resource._http.patch.call_args[1]["json"] + assert body["isDefault"] is True + + def test_update_sends_not_found_html(self): + resource = _make_resource() + resource.update("links.example.com", not_found_html="

404

") + body = resource._http.patch.call_args[1]["json"] + assert body["notFoundHtml"] == "

404

" + + def test_update_returns_custom_domain(self): + resource = _make_resource() + result = resource.update("links.example.com", is_default=True) + assert isinstance(result, CustomDomain) + + def test_remove_calls_delete(self): + resource = _make_resource() + resource.remove("links.example.com") + resource._http.delete.assert_called_once_with( + "/api/user/domains/links.example.com" + ) + + def test_check_calls_correct_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = {"available": True} + resource.check("links.example.com") + resource._http.get.assert_called_once_with( + "/api/domains/check/links.example.com" + ) diff --git a/tests/test_data_export.py b/tests/test_data_export.py new file mode 100644 index 0000000..39a3100 --- /dev/null +++ b/tests/test_data_export.py @@ -0,0 +1,39 @@ +"""Unit tests for the Data Export resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.resources.data_export import DataExportResource + + +def _make_resource(text="short_code,long_url\nabc,https://example.com"): + http = MagicMock() + http.get_text.return_value = text + return DataExportResource(http) + + +class TestDataExport: + def test_export_links_calls_correct_endpoint(self): + resource = _make_resource() + resource.export_links() + resource._http.get_text.assert_called_once_with("/api/export/links") + + def test_export_links_returns_string(self): + resource = _make_resource("col1,col2\nval1,val2") + result = resource.export_links() + assert isinstance(result, str) + assert "col1" in result + + def test_export_link_stats_calls_correct_endpoint(self): + resource = _make_resource() + resource.export_link_stats("abc123") + resource._http.get_text.assert_called_once_with("/api/export/stats/abc123") + + def test_export_link_stats_returns_string(self): + resource = _make_resource("date,clicks\n2026-01-01,42") + result = resource.export_link_stats("abc123") + assert isinstance(result, str) + assert "clicks" in result diff --git a/tests/test_folders.py b/tests/test_folders.py index 8f3bca3..5c1e35f 100644 --- a/tests/test_folders.py +++ b/tests/test_folders.py @@ -1,9 +1,78 @@ -"""Integration tests for the Folders resource.""" +"""Tests for the Folders resource.""" + +from __future__ import annotations import time +from unittest.mock import MagicMock from awsysco import Client from awsysco.models import Folder, FolderList +from awsysco.resources.folders import FoldersResource + + +# --------------------------------------------------------------------------- +# Unit tests — no network required +# --------------------------------------------------------------------------- + +_FOLDER_DATA = { + "id": "folder1", + "name": "Test Folder", + "color": "#FF5733", + "linkCount": 0, + "createdAt": "2026-01-01T00:00:00Z", +} + + +def _make_resource(): + http = MagicMock() + http.get.return_value = {"folders": [_FOLDER_DATA], "limit": 10, "used": 1} + http.post.return_value = _FOLDER_DATA + http.patch.return_value = _FOLDER_DATA + http.delete.return_value = None + return FoldersResource(http) + + +class TestFoldersUnit: + def test_update_calls_patch_endpoint(self): + resource = _make_resource() + resource.update("folder1", name="Renamed") + resource._http.patch.assert_called_once_with( + "/api/v1/folders/folder1", json={"name": "Renamed"} + ) + + def test_update_sends_color(self): + resource = _make_resource() + resource.update("folder1", color="#0000FF") + body = resource._http.patch.call_args[1]["json"] + assert body["color"] == "#0000FF" + + def test_update_returns_folder(self): + resource = _make_resource() + result = resource.update("folder1", name="New Name") + assert isinstance(result, Folder) + + def test_update_omits_none_fields(self): + resource = _make_resource() + resource.update("folder1", name="Name Only") + body = resource._http.patch.call_args[1]["json"] + assert "color" not in body + + def test_create_calls_endpoint(self): + resource = _make_resource() + resource.create("My Folder", color="#AABBCC") + body = resource._http.post.call_args[1]["json"] + assert body["name"] == "My Folder" + assert body["color"] == "#AABBCC" + + def test_list_calls_endpoint(self): + resource = _make_resource() + resource.list() + resource._http.get.assert_called_once_with("/api/v1/folders") + + +# --------------------------------------------------------------------------- +# Integration tests — require AWSYS_API_KEY +# --------------------------------------------------------------------------- def _folder_name() -> str: @@ -19,7 +88,6 @@ def test_create_folder(self, client: Client) -> None: folder = client.folders.create(_folder_name(), color="#FF5733") assert isinstance(folder, Folder) assert folder.id is not None - # Clean up client.folders.delete(folder.id) def test_list_contains_created_folder(self, client: Client) -> None: @@ -31,8 +99,6 @@ def test_list_contains_created_folder(self, client: Client) -> None: assert isinstance(folder_list, FolderList) ids = [f.id for f in folder_list.folders] assert folder.id in ids - - # Clean up client.folders.delete(folder.id) def test_assign_link_to_folder(self, client: Client) -> None: @@ -42,10 +108,8 @@ def test_assign_link_to_folder(self, client: Client) -> None: link = client.links.create(_unique_url()) assert link.short_code is not None - # Should not raise client.folders.assign_link(link.short_code, folder.id) - # Clean up client.folders.delete(folder.id) client.links.delete(link.short_code) @@ -57,16 +121,12 @@ def test_remove_link_from_folder(self, client: Client) -> None: assert link.short_code is not None client.folders.assign_link(link.short_code, folder.id) - # Remove — should not raise client.folders.remove_link(link.short_code) - # Clean up client.folders.delete(folder.id) client.links.delete(link.short_code) def test_delete_folder(self, client: Client) -> None: folder = client.folders.create(_folder_name()) assert folder.id is not None - - # Should not raise client.folders.delete(folder.id) diff --git a/tests/test_links.py b/tests/test_links.py index c937d3f..68d6ea7 100644 --- a/tests/test_links.py +++ b/tests/test_links.py @@ -1,11 +1,126 @@ -"""Integration tests for the Links resource.""" +"""Tests for the Links resource.""" + +from __future__ import annotations import time +from unittest.mock import MagicMock import pytest from awsysco import Client, AwsysNotFoundError from awsysco.models import Link, LinkList +from awsysco.resources.links import LinksResource + + +# --------------------------------------------------------------------------- +# Unit tests — no network required +# --------------------------------------------------------------------------- + +_LINK_DATA = { + "id": "link1", + "shortUrl": "https://awsys.co/abc", + "shortCode": "abc", + "long": "https://example.com", + "clicks": 0, +} + + +def _make_resource(return_value=None): + http = MagicMock() + http.post.return_value = return_value or _LINK_DATA + http.get.return_value = return_value or _LINK_DATA + http.patch.return_value = return_value or _LINK_DATA + http.delete.return_value = None + return LinksResource(http) + + +class TestLinksCreateUnit: + def test_create_sends_url(self): + resource = _make_resource() + resource.create("https://example.com") + body = resource._http.post.call_args[1]["json"] + assert body["url"] == "https://example.com" + + def test_create_sends_custom_slug(self): + resource = _make_resource() + resource.create("https://example.com", custom_slug="myslug") + body = resource._http.post.call_args[1]["json"] + assert body["customSlug"] == "myslug" + + def test_create_sends_routing_rules(self): + resource = _make_resource() + rules = [{"country": "US", "redirectUrl": "https://us.example.com"}] + resource.create("https://example.com", routing_rules=rules) + body = resource._http.post.call_args[1]["json"] + assert body["routingRules"] == rules + + def test_create_sends_og_meta(self): + resource = _make_resource() + og = {"title": "My Title", "description": "desc"} + resource.create("https://example.com", og_meta=og) + body = resource._http.post.call_args[1]["json"] + assert body["ogMeta"] == og + + def test_create_sends_geo_restriction(self): + resource = _make_resource() + geo = {"allowedCountries": ["US", "CA"]} + resource.create("https://example.com", geo_restriction=geo) + body = resource._http.post.call_args[1]["json"] + assert body["geoRestriction"] == geo + + def test_create_sends_password(self): + resource = _make_resource() + resource.create("https://example.com", password="secret") + body = resource._http.post.call_args[1]["json"] + assert body["password"] == "secret" + + def test_create_sends_pass_ad_click_ids(self): + resource = _make_resource() + resource.create("https://example.com", pass_ad_click_ids=True) + body = resource._http.post.call_args[1]["json"] + assert body["passAdClickIds"] is True + + def test_create_sends_folder_id(self): + resource = _make_resource() + resource.create("https://example.com", folder_id="folder1") + body = resource._http.post.call_args[1]["json"] + assert body["folderId"] == "folder1" + + def test_create_sends_tags(self): + resource = _make_resource() + resource.create("https://example.com", tags=["promo", "social"]) + body = resource._http.post.call_args[1]["json"] + assert body["tags"] == ["promo", "social"] + + def test_create_returns_link(self): + resource = _make_resource() + result = resource.create("https://example.com") + assert isinstance(result, Link) + + def test_update_sends_url(self): + resource = _make_resource() + resource.update("abc", url="https://new.example.com") + body = resource._http.patch.call_args[1]["json"] + assert body["url"] == "https://new.example.com" + + def test_update_sends_tags(self): + resource = _make_resource() + resource.update("abc", tags=["new-tag"]) + body = resource._http.patch.call_args[1]["json"] + assert body["tags"] == ["new-tag"] + + def test_update_omits_none_fields(self): + resource = _make_resource() + resource.update("abc", max_clicks=50) + body = resource._http.patch.call_args[1]["json"] + assert "url" not in body + assert "tags" not in body + assert body["maxClicks"] == 50 + + +# --------------------------------------------------------------------------- +# Integration tests — require AWSYS_API_KEY +# --------------------------------------------------------------------------- def _unique_url() -> str: @@ -38,7 +153,6 @@ def test_list_default_limit(self, client: Client) -> None: class TestLinksGet: def test_get_returns_correct_link(self, client: Client) -> None: - # Create a link first created = client.links.create(_unique_url()) short_code = created.short_code assert short_code is not None diff --git a/tests/test_namespace.py b/tests/test_namespace.py new file mode 100644 index 0000000..6f41cc5 --- /dev/null +++ b/tests/test_namespace.py @@ -0,0 +1,84 @@ +"""Unit tests for the Namespace resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.models import NamespaceCheckResult, NamespaceInfo +from awsysco.resources.namespace import NamespaceResource + + +def _make_resource(): + http = MagicMock() + return NamespaceResource(http) + + +_NAMESPACE_INFO = { + "hasAccess": True, + "namespace": "myco", + "tier": "pro", + "upgradeRequired": False, +} + +_CHECK_RESULT = { + "namespace": "myco", + "available": True, + "reason": None, + "previewUrl": "https://awsys.co/myco/test", +} + + +class TestNamespace: + def test_get_calls_correct_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = _NAMESPACE_INFO + resource.get() + resource._http.get.assert_called_once_with("/api/user/namespace") + + def test_get_returns_namespace_info(self): + resource = _make_resource() + resource._http.get.return_value = _NAMESPACE_INFO + result = resource.get() + assert isinstance(result, NamespaceInfo) + assert result.namespace == "myco" + + def test_check_calls_correct_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = _CHECK_RESULT + resource.check("myco") + resource._http.get.assert_called_once_with("/api/namespace/check/myco") + + def test_check_returns_check_result(self): + resource = _make_resource() + resource._http.get.return_value = _CHECK_RESULT + result = resource.check("myco") + assert isinstance(result, NamespaceCheckResult) + assert result.available is True + + def test_claim_calls_correct_endpoint(self): + resource = _make_resource() + resource._http.post.return_value = _NAMESPACE_INFO + resource.claim("myco") + resource._http.post.assert_called_once_with( + "/api/user/namespace", json={"namespace": "myco"} + ) + + def test_claim_returns_namespace_info(self): + resource = _make_resource() + resource._http.post.return_value = _NAMESPACE_INFO + result = resource.claim("myco") + assert isinstance(result, NamespaceInfo) + + def test_release_calls_delete(self): + resource = _make_resource() + resource._http.delete.return_value = None + resource.release() + resource._http.delete.assert_called_once_with("/api/user/namespace") + + def test_release_returns_dict(self): + resource = _make_resource() + resource._http.delete.return_value = None + result = resource.release() + assert isinstance(result, dict) diff --git a/tests/test_saved_views.py b/tests/test_saved_views.py new file mode 100644 index 0000000..9269e78 --- /dev/null +++ b/tests/test_saved_views.py @@ -0,0 +1,83 @@ +"""Unit tests for the Saved Views resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.models import SavedView +from awsysco.resources.saved_views import SavedViewsResource + + +_VIEW_DATA = { + "id": "v1", + "name": "My View", + "filters": {"tag": "promo", "status": "active"}, + "createdAt": "2026-01-01T00:00:00Z", +} + + +def _make_resource(): + http = MagicMock() + http.get.return_value = [_VIEW_DATA] + http.post.return_value = _VIEW_DATA + http.patch.return_value = _VIEW_DATA + http.delete.return_value = None + return SavedViewsResource(http) + + +class TestSavedViews: + def test_list_calls_endpoint(self): + resource = _make_resource() + resource.list() + resource._http.get.assert_called_once_with("/api/views") + + def test_list_returns_list_of_saved_views(self): + resource = _make_resource() + result = resource.list() + assert isinstance(result, list) + assert all(isinstance(v, SavedView) for v in result) + + def test_list_handles_dict_wrapper(self): + resource = _make_resource() + resource._http.get.return_value = {"views": [_VIEW_DATA]} + result = resource.list() + assert len(result) == 1 + assert isinstance(result[0], SavedView) + + def test_create_calls_endpoint(self): + resource = _make_resource() + resource.create("My View", {"tag": "promo"}) + resource._http.post.assert_called_once_with( + "/api/views", json={"name": "My View", "filters": {"tag": "promo"}} + ) + + def test_create_returns_saved_view(self): + resource = _make_resource() + result = resource.create("My View", {}) + assert isinstance(result, SavedView) + assert result.id == "v1" + + def test_update_calls_patch(self): + resource = _make_resource() + resource.update("v1", name="Renamed") + resource._http.patch.assert_called_once_with( + "/api/views/v1", json={"name": "Renamed"} + ) + + def test_update_with_filters(self): + resource = _make_resource() + resource.update("v1", filters={"status": "inactive"}) + body = resource._http.patch.call_args[1]["json"] + assert body["filters"] == {"status": "inactive"} + + def test_update_returns_saved_view(self): + resource = _make_resource() + result = resource.update("v1", name="Renamed") + assert isinstance(result, SavedView) + + def test_delete_calls_endpoint(self): + resource = _make_resource() + resource.delete("v1") + resource._http.delete.assert_called_once_with("/api/views/v1") diff --git a/tests/test_tags.py b/tests/test_tags.py new file mode 100644 index 0000000..a6c1bdd --- /dev/null +++ b/tests/test_tags.py @@ -0,0 +1,65 @@ +"""Unit tests for the Tags resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from awsysco.resources.tags import TagsResource + + +def _make_resource(return_value=None): + http = MagicMock() + http.post.return_value = return_value or {"ok": True} + http.delete.return_value = return_value or {"ok": True} + return TagsResource(http) + + +class TestTagsAdd: + def test_add_calls_correct_endpoint(self): + resource = _make_resource() + resource.add("abc123", "promo") + resource._http.post.assert_called_once_with( + "/api/link/abc123/tags", json={"tag": "promo"} + ) + + def test_add_returns_dict(self): + resource = _make_resource({"tag": "promo", "ok": True}) + result = resource.add("abc123", "promo") + assert isinstance(result, dict) + + def test_add_encodes_short_path(self): + resource = _make_resource() + resource.add("ns/slug", "test") + resource._http.post.assert_called_once_with( + "/api/link/ns%2Fslug/tags", json={"tag": "test"} + ) + + def test_add_returns_empty_dict_on_none(self): + http = MagicMock() + http.post.return_value = None + resource = TagsResource(http) + result = resource.add("abc", "tag") + assert result == {} + + +class TestTagsRemove: + def test_remove_calls_correct_endpoint(self): + resource = _make_resource() + resource.remove("abc123", "promo") + resource._http.delete.assert_called_once_with( + "/api/link/abc123/tags/promo" + ) + + def test_remove_encodes_tag(self): + resource = _make_resource() + resource.remove("abc123", "hello world") + resource._http.delete.assert_called_once_with( + "/api/link/abc123/tags/hello%20world" + ) + + def test_remove_returns_dict(self): + resource = _make_resource({"removed": True}) + result = resource.remove("abc123", "promo") + assert isinstance(result, dict) diff --git a/tests/test_trust_score.py b/tests/test_trust_score.py new file mode 100644 index 0000000..554fa02 --- /dev/null +++ b/tests/test_trust_score.py @@ -0,0 +1,55 @@ +"""Unit tests for the Trust Score resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.models import TrustScoreResult +from awsysco.resources.trust_score import TrustScoreResource + + +def _make_resource(return_value=None): + http = MagicMock() + http.get.return_value = return_value or { + "short": "abc123", + "long": "https://example.com", + "score": 0.95, + "status": "safe", + "threats": [], + "scannedAt": "2026-01-01T00:00:00Z", + } + return TrustScoreResource(http) + + +class TestTrustScore: + def test_scan_calls_correct_endpoint(self): + resource = _make_resource() + resource.scan("abc123") + resource._http.get.assert_called_once_with("/api/link-scan/abc123") + + def test_scan_encodes_slash(self): + resource = _make_resource() + resource.scan("ns/slug") + resource._http.get.assert_called_once_with("/api/link-scan/ns%2Fslug") + + def test_scan_returns_trust_score_result(self): + resource = _make_resource() + result = resource.scan("abc123") + assert isinstance(result, TrustScoreResult) + + def test_scan_populates_score(self): + resource = _make_resource() + result = resource.scan("abc123") + assert result.score == 0.95 + + def test_scan_populates_status(self): + resource = _make_resource() + result = resource.scan("abc123") + assert result.status == "safe" + + def test_scan_populates_threats(self): + resource = _make_resource() + result = resource.scan("abc123") + assert result.threats == [] diff --git a/tests/test_utm_templates.py b/tests/test_utm_templates.py new file mode 100644 index 0000000..820bedb --- /dev/null +++ b/tests/test_utm_templates.py @@ -0,0 +1,73 @@ +"""Unit tests for the UTM Templates resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.models import UtmTemplate +from awsysco.resources.utm_templates import UtmTemplatesResource + + +def _make_resource(me_response=None): + http = MagicMock() + if me_response is not None: + http.get.return_value = me_response + else: + http.get.return_value = { + "uid": "user1", + "utmTemplates": [ + {"id": "t1", "name": "Google Ads", "source": "google", "medium": "cpc", "campaign": "brand"}, + ], + } + http.post.return_value = {"id": "t2", "name": "New"} + http.delete.return_value = None + return UtmTemplatesResource(http) + + +class TestUtmTemplates: + def test_list_calls_me_endpoint(self): + resource = _make_resource() + resource.list() + resource._http.get.assert_called_once_with("/api/v1/me") + + def test_list_returns_utm_templates(self): + resource = _make_resource() + result = resource.list() + assert isinstance(result, list) + assert all(isinstance(t, UtmTemplate) for t in result) + + def test_list_returns_empty_when_no_templates(self): + resource = _make_resource(me_response={"uid": "user1"}) + result = resource.list() + assert result == [] + + def test_list_parses_template_fields(self): + resource = _make_resource() + result = resource.list() + assert result[0].name == "Google Ads" + assert result[0].source == "google" + + def test_create_calls_correct_endpoint(self): + resource = _make_resource() + resource.create("My Template", "google", "cpc", "brand") + resource._http.post.assert_called_once() + call_args = resource._http.post.call_args + assert call_args[0][0] == "/api/user/utm-templates" + + def test_create_sends_correct_body(self): + resource = _make_resource() + resource.create("My Template", "google", "cpc", "brand", term="shoes", content="banner") + body = resource._http.post.call_args[1]["json"] + assert body["name"] == "My Template" + assert body["source"] == "google" + assert body["medium"] == "cpc" + assert body["campaign"] == "brand" + assert body["term"] == "shoes" + assert body["content"] == "banner" + + def test_delete_calls_correct_endpoint(self): + resource = _make_resource() + resource.delete("t1") + resource._http.delete.assert_called_once_with("/api/user/utm-templates/t1") diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py new file mode 100644 index 0000000..5539061 --- /dev/null +++ b/tests/test_webhooks.py @@ -0,0 +1,94 @@ +"""Unit tests for the Webhooks resource.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from awsysco.models import Webhook +from awsysco.resources.webhooks import WebhooksResource + + +_WEBHOOK_DATA = { + "id": "wh1", + "url": "https://example.com/hook", + "events": ["link.created", "link.click"], + "name": "My Webhook", + "enabled": True, + "createdAt": "2026-01-01T00:00:00Z", +} + + +def _make_resource(): + http = MagicMock() + http.get.return_value = {"webhooks": [_WEBHOOK_DATA]} + http.post.return_value = _WEBHOOK_DATA + http.patch.return_value = _WEBHOOK_DATA + http.delete.return_value = None + return WebhooksResource(http) + + +class TestWebhooks: + def test_list_event_types_calls_endpoint(self): + resource = _make_resource() + resource._http.get.return_value = {"eventTypes": ["link.created"]} + resource.list_event_types() + resource._http.get.assert_called_once_with("/api/webhooks/event-types") + + def test_list_calls_endpoint(self): + resource = _make_resource() + resource.list() + resource._http.get.assert_called_once_with("/api/webhooks") + + def test_create_returns_webhook(self): + resource = _make_resource() + result = resource.create("https://example.com/hook", ["link.created"]) + assert isinstance(result, Webhook) + assert result.id == "wh1" + + def test_create_sends_correct_body(self): + resource = _make_resource() + resource.create( + "https://example.com/hook", + ["link.created"], + name="My Webhook", + secret="s3cr3t", + ) + body = resource._http.post.call_args[1]["json"] + assert body["url"] == "https://example.com/hook" + assert body["events"] == ["link.created"] + assert body["name"] == "My Webhook" + assert body["secret"] == "s3cr3t" + + def test_create_without_optional_fields(self): + resource = _make_resource() + resource.create("https://example.com/hook", ["link.click"]) + body = resource._http.post.call_args[1]["json"] + assert "name" not in body + assert "secret" not in body + + def test_update_calls_patch(self): + resource = _make_resource() + resource.update("wh1", enabled=False) + resource._http.patch.assert_called_once_with( + "/api/webhooks/wh1", json={"enabled": False} + ) + + def test_update_returns_webhook(self): + resource = _make_resource() + result = resource.update("wh1", name="Renamed") + assert isinstance(result, Webhook) + + def test_delete_calls_endpoint(self): + resource = _make_resource() + resource.delete("wh1") + resource._http.delete.assert_called_once_with("/api/webhooks/wh1") + + def test_test_calls_correct_endpoint(self): + resource = _make_resource() + resource._http.post.return_value = {"sent": True} + resource.test("wh1", "link.created") + resource._http.post.assert_called_once_with( + "/api/webhooks/wh1/test", json={"eventType": "link.created"} + )