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

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions sdks/python1/pyproject.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
[project]
name = "sint-sdk"
version = "0.1.0"
description = "Python SDK for Sint Protocol Gateway"
dependencies = [
"httpx>=0.27.0",
]

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build-meta"
Empty file added sdks/python1/sint/__init__.py
Empty file.
54 changes: 54 additions & 0 deletions sdks/python1/sint/client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
import httpx
from typing import Any, Dict, List, Optional
from .errors import SintError
from .types import PolicyDecision, SintCapabilityToken

class SintClient:
def __init__(self, base_url: str, api_key: str, timeout: float = 10.0):
self.base_url = base_url.rstrip("/")
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.timeout = timeout

async def _request(self, method: str, path: str, data: Optional[Dict] = None) -> Any:
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.request(
method,
f"{self.base_url}{path}",
json=data,
headers=self.headers
)
if not response.is_success:
try:
error_data = response.json()
raise SintError(
response.status_code,
error_data.get("code", "UNKNOWN"),
error_data.get("message", "An error occurred")
)
except Exception:
raise SintError(response.status_code, "ERROR", response.text)
return response.json()

async def intercept(self, request_data: Dict[str, Any]) -> PolicyDecision:
return await self._request("POST", "/intercept", data=request_data)

async def issue_token(self, params: Dict[str, Any]) -> SintCapabilityToken:
return await self._request("POST", "/tokens", data=params)

async def revoke_token(self, token_id: str) -> None:
await self._request("DELETE", f"/tokens/{token_id}")

async def get_ledger(self, filters: Optional[Dict[str, Any]] = None) -> List[Any]:
return await self._request("GET", "/ledger", data=filters)

async def pending_approvals(self) -> List[Any]:
return await self._request("GET", "/approvals/pending")

async def resolve_approval(self, request_id: str, approved: bool) -> Any:
return await self._request("POST", f"/approvals/{request_id}/resolve", data={"approved": approved})

async def delegate_token(self, parent_token_id: str, params: Dict[str, Any]) -> Any:
return await self._request("POST", f"/tokens/{parent_token_id}/delegate", data=params)
7 changes: 7 additions & 0 deletions sdks/python1/sint/errors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
class SintError(Exception):
"""Custom error for Sint SDK non-2xx responses."""
def __init__(self, status_code: int, code: str, message: str):
self.status_code = status_code
self.code = code
self.message = message
super().__init__(f"[{status_code}] {code}: {message}")
9 changes: 9 additions & 0 deletions sdks/python1/sint/types.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
from typing import TypedDict, List, Optional, Any

class PolicyDecision(TypedDict):
decision: str
reason: Optional[str]

class SintCapabilityToken(TypedDict):
token: str
expires_at: str
15 changes: 15 additions & 0 deletions sdks/python1/tests/test_client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
import pytest
from sint.client import SintClient

@pytest.mark.asyncio
async def test_intercept(httpx_mock):
httpx_mock.add_response(
method="POST",
url="https://api.example.com/intercept",
json={"decision": "PERMIT", "reason": "OK"},
status_code=200
)

client = SintClient("https://api.example.com", "test-key")
result = await client.intercept({"action": "read"})
assert result["decision"] == "PERMIT"
Loading