feat: Introduce AsyncGavaConnect and CheckersClient for asynchronous PIN validation

- Added AsyncGavaConnect class for async operations with checkers API.
- Implemented CheckersClient for KRA PIN validation with methods for validating PINs.
- Created BasicTokenEndpointProvider for handling token retrieval with Basic auth.
- Introduced BasicPair data class for managing client credentials.
- Updated auth module to include new classes and methods.
- Added smoke tests and unit tests for validation and error handling.
- Updated dependencies in pyproject.toml to include pydantic.
- Enhanced error handling for API responses, including rate limiting and unauthorized access.

Author: musale

README.md

@@ -40,11 +40,74 @@ uv install
 
 ## Quick Start
 
+### Basic SDK Usage
+
+```python
+import gavaconnect
+
+# Create configuration
+config = gavaconnect.SDKConfig(base_url="https://sbx.kra.go.ke")
+
+# Work with errors and basic types
+try:
+    # Your API calls here
+    pass
+except gavaconnect.APIError as e:
+    print(f"API Error: {e.status} - {e.message}")
+```
+
+### PIN Validation (requires httpx and pydantic)
+
 ```python
-from gavaconnect import main
+from gavaconnect.facade_async import AsyncGavaConnect
+from gavaconnect.config import SDKConfig
+
+async def validate_pin():
+    config = SDKConfig(base_url="https://sbx.kra.go.ke")
+
+    async with AsyncGavaConnect(
+        config,
+        checkers_client_id="your_client_id",
+        checkers_client_secret="your_client_secret"
+    ) as sdk:
+        result = await sdk.checkers.validate_pin(pin="A000000000B")
+        print(result.model_dump(by_alias=True))
+        # Output: {"PIN": "A000000000B", "TaxPayerName": "...", "status": "VALID", "valid": true}
+
+# Run the async function
+import asyncio
+asyncio.run(validate_pin())
+```
 
-# Basic usage example
-main()
+### Advanced Usage
+
+```python
+from gavaconnect.resources.checkers import CheckersClient, PinCheckResult
+from gavaconnect.auth import BasicTokenEndpointProvider, BasicPair, BearerAuthPolicy
+from gavaconnect.http.transport import AsyncTransport
+
+# Manual client setup for advanced use cases
+async def advanced_usage():
+    config = SDKConfig(base_url="https://sbx.kra.go.ke")
+    transport = AsyncTransport(config)
+
+    # Setup authentication
+    provider = BasicTokenEndpointProvider(
+        token_url="https://sbx.kra.go.ke/v1/token/generate",
+        basic=BasicPair("client_id", "client_secret"),
+        method="GET"
+    )
+    auth = BearerAuthPolicy(provider)
+
+    # Create client
+    client = CheckersClient(transport, auth)
+
+    # Use different validation methods
+    result1 = await client.validate_pin(pin="A000000000B")
+    result2 = await client.validate_pin_get(pin="A000000000B", query_key="PIN")
+    result3 = await client.validate_pin_raw({"PIN": "A000000000B", "extra": "data"})
+
+    await transport.close()
 ```
 
 ## Development

gavaconnect/__init__.py

@@ -5,6 +5,10 @@
 from .config import SDKConfig
 from .errors import APIError, RateLimitError, SDKError, TransportError
 
+# Note: AsyncGavaConnect and CheckersClient require httpx and pydantic
+# Import them explicitly: from gavaconnect.facade_async import AsyncGavaConnect
+# or from gavaconnect.resources.checkers import CheckersClient
+
 __all__ = [
     "__version__",
     "SDKConfig",

gavaconnect/auth/__init__.py

@@ -2,13 +2,16 @@
 
 from .basic import BasicAuthPolicy, BasicCredentials
 from .bearer import AuthPolicy, BearerAuthPolicy, TokenProvider
-from .providers import ClientCredentialsProvider
+from .credentials import BasicPair
+from .providers import BasicTokenEndpointProvider, ClientCredentialsProvider
 
 __all__ = [
     "AuthPolicy",
-    "BasicAuthPolicy",
+    "BasicAuthPolicy", 
     "BasicCredentials",
+    "BasicPair",
     "BearerAuthPolicy",
     "TokenProvider",
+    "BasicTokenEndpointProvider",
     "ClientCredentialsProvider",
 ]

gavaconnect/auth/credentials.py

@@ -0,0 +1,11 @@
+"""Basic credential types for authentication."""
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class BasicPair:
+    """Basic auth credential pair for token endpoints."""
+
+    client_id: str
+    client_secret: str

gavaconnect/auth/providers.py

@@ -2,9 +2,12 @@
 
 import asyncio
 import time
+from typing import Literal
 
 import httpx
 
+from .credentials import BasicPair
+
 # Minimum token TTL to prevent rapid refresh cycles
 MIN_TOKEN_TTL_S = 30.0
 
@@ -86,3 +89,74 @@ async def refresh(self) -> str:
         async with self._lock:
             self._token, self._exp = await self._fetch()
             return self._token
+
+
+class BasicTokenEndpointProvider:
+    """Token provider using HTTP Basic auth against a token endpoint."""
+
+    def __init__(
+        self,
+        token_url: str,
+        basic: BasicPair,
+        method: Literal["GET", "POST"] = "POST",
+        early_refresh_s: int = 60,
+        client: httpx.AsyncClient | None = None,
+        token_timeout_s: float = 10.0,
+    ) -> None:
+        """Initialize the basic token endpoint provider.
+
+        Args:
+            token_url: Token endpoint URL.
+            basic: Basic auth credentials.
+            method: HTTP method to use (GET or POST).
+            early_refresh_s: Seconds before expiry to refresh token.
+            client: Optional HTTP client to use.
+            token_timeout_s: Timeout for token requests in seconds.
+
+        """
+        self._url = token_url
+        self._basic = basic
+        self._method = method
+        self._early = early_refresh_s
+        self._client = client or httpx.AsyncClient(timeout=token_timeout_s)
+        self._lock = asyncio.Lock()
+        # Security note: tokens stored in memory - consider using keyring for production
+        self._token, self._exp = "", 0.0
+
+    async def _fetch(self) -> tuple[str, float]:
+        """Fetch a new token from the endpoint."""
+        auth = (self._basic.client_id, self._basic.client_secret)
+        
+        if self._method == "GET":
+            resp = await self._client.get(self._url, auth=auth)
+        else:
+            resp = await self._client.post(self._url, auth=auth)
+            
+        resp.raise_for_status()
+        payload = resp.json()
+        ttl = float(payload.get("expires_in", 3600))
+        return payload["access_token"], time.time() + max(MIN_TOKEN_TTL_S, ttl - self._early)
+
+    async def get_token(self) -> str:
+        """Get the current access token, refreshing if necessary.
+
+        Returns:
+            The access token.
+
+        """
+        async with self._lock:
+            if self._token and time.time() < self._exp:
+                return self._token
+            self._token, self._exp = await self._fetch()
+            return self._token
+
+    async def refresh(self) -> str:
+        """Force refresh the access token.
+
+        Returns:
+            The new access token.
+
+        """
+        async with self._lock:
+            self._token, self._exp = await self._fetch()
+            return self._token

gavaconnect/facade_async.py

@@ -0,0 +1,60 @@
+"""Async facade for GavaConnect SDK."""
+
+from types import TracebackType
+
+from gavaconnect.auth import BasicPair, BasicTokenEndpointProvider, BearerAuthPolicy
+from gavaconnect.config import SDKConfig
+from gavaconnect.http.transport import AsyncTransport
+from gavaconnect.resources.checkers import CheckersClient
+
+__all__ = ["AsyncGavaConnect"]
+
+
+class AsyncGavaConnect:
+    """Async facade for GavaConnect SDK with per-family credentials."""
+
+    def __init__(
+        self,
+        config: SDKConfig,
+        *,
+        checkers_client_id: str,
+        checkers_client_secret: str,
+        token_url: str = "https://sbx.kra.go.ke/v1/token/generate",
+    ) -> None:
+        """Initialize the async GavaConnect client.
+
+        Args:
+            config: SDK configuration.
+            checkers_client_id: Client ID for checkers API.
+            checkers_client_secret: Client secret for checkers API.
+            token_url: Token endpoint URL.
+
+        """
+        self._config = config
+        self._tr = AsyncTransport(config)
+        
+        # Setup checkers client with Basic -> Bearer flow
+        provider = BasicTokenEndpointProvider(
+            token_url=token_url,
+            basic=BasicPair(checkers_client_id, checkers_client_secret),
+            method="GET",
+            early_refresh_s=60,
+        )
+        self.checkers = CheckersClient(self._tr, BearerAuthPolicy(provider))
+
+    async def __aenter__(self) -> "AsyncGavaConnect":
+        """Async context manager entry."""
+        return self
+
+    async def __aexit__(
+        self, 
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None, 
+        exc_tb: TracebackType | None,
+    ) -> None:
+        """Async context manager exit."""
+        await self.close()
+
+    async def close(self) -> None:
+        """Close the client and cleanup resources."""
+        await self._tr.close()

gavaconnect/resources/__init__.py

@@ -0,0 +1,6 @@
+"""Resources package for GavaConnect SDK."""
+
+# Checkers resources require httpx and pydantic dependencies
+# Import explicitly: from gavaconnect.resources.checkers import CheckersClient
+
+__all__ = []  # Explicit imports required due to dependencies

gavaconnect/resources/checkers/__init__.py

@@ -0,0 +1,5 @@
+"""Checkers resource package."""
+
+from ._pin import CheckersClient, PinCheckResult
+
+__all__ = ["CheckersClient", "PinCheckResult"]

gavaconnect/resources/checkers/_pin.py

@@ -0,0 +1,90 @@
+"""PIN validation client for KRA checkers."""
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from gavaconnect.auth.bearer import BearerAuthPolicy
+from gavaconnect.helpers.idempotency import idempotency_headers
+from gavaconnect.http.transport import AsyncTransport
+
+
+class PinCheckResult(BaseModel):
+    """Result of PIN validation check."""
+
+    pin: str | None = Field(default=None, alias="PIN")
+    taxpayer_name: str | None = Field(default=None, alias="TaxPayerName")
+    status: str | None = None
+    valid: bool | None = None
+    
+    model_config = ConfigDict(populate_by_name=True, extra="allow")
+
+
+class CheckersClient:
+    """Client for KRA PIN validation endpoints."""
+
+    def __init__(self, tr: AsyncTransport, auth: BearerAuthPolicy) -> None:
+        """Initialize the checkers client.
+
+        Args:
+            tr: Transport instance for HTTP requests.
+            auth: Bearer authentication policy.
+
+        """
+        self._tr = tr
+        self._auth = auth
+
+    async def validate_pin(self, *, pin: str, pin_key: str = "PIN") -> PinCheckResult:
+        """Validate a KRA PIN using POST with JSON payload.
+
+        Args:
+            pin: The PIN to validate.
+            pin_key: The JSON key name for the PIN field.
+
+        Returns:
+            PIN validation result.
+
+        """
+        payload = {pin_key: pin}
+        return await self.validate_pin_raw(payload)
+
+    async def validate_pin_get(self, *, pin: str, query_key: str = "PIN") -> PinCheckResult:
+        """Validate a KRA PIN using GET with query parameters.
+
+        Args:
+            pin: The PIN to validate.
+            query_key: The query parameter name for the PIN field.
+
+        Returns:
+            PIN validation result.
+
+        """
+        resp = await self._tr.request(
+            "GET",
+            "/checker/v1/pinbypin",
+            auth=self._auth,
+            params={query_key: pin},
+        )
+        self._tr.raise_for_api_error(resp)
+        return PinCheckResult.model_validate(resp.json())
+
+    async def validate_pin_raw(self, payload: dict[str, Any]) -> PinCheckResult:
+        """Validate a PIN using raw payload.
+
+        Args:
+            payload: Raw JSON payload to send.
+
+        Returns:
+            PIN validation result.
+
+        """
+        headers = idempotency_headers()  # Make POST requests retryable
+        resp = await self._tr.request(
+            "POST", 
+            "/checker/v1/pinbypin",
+            auth=self._auth,
+            json=payload,
+            headers=headers,
+        )
+        self._tr.raise_for_api_error(resp)
+        return PinCheckResult.model_validate(resp.json())

pyproject.toml

@@ -10,6 +10,7 @@ authors = [
 requires-python = ">=3.13"
 dependencies = [
     "httpx>=0.28.1",
+    "pydantic>=2.0.0",
 ]
 license = { text = "MIT" }
 keywords = ["gavaconnect", "sdk", "api"]