From cbaf03857b4d5973866e4aebd5f20539de0193b4 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 15:06:03 +0000
Subject: [PATCH 1/3] feat(connectors): add Google Sheet connector (Week 2
 roadmap)

Second connector per MAGIC_MONTH1_TASK_LIST.md (magic-business): reads
and writes Order/Customer data in a Google Sheet, for shops without a
dedicated order-management system yet.

- Auth via Google service account (JWT bearer through google-auth);
  the actual Sheets API v4 calls go through httpx like the Zalo
  connector, so credential refresh uses a small httpx-based adapter
  for google.auth.transport.Request instead of pulling in the separate
  `requests` library just for that one call.
- Operations: read_range, update_range, append_row, batch_update
- Common Schema mapping is header-name-based (Customer/Order), with
  Order.items round-tripped as a JSON string in an items_json column
  since sheets have no nested types
- Error mapping follows Google's standard {error: {code, status,
  message}} shape: RESOURCE_EXHAUSTED -> RateLimitError, 5xx/UNAVAILABLE/
  INTERNAL -> TransientError (retried), everything else -> PermanentError
- 13 tests (auth mocked, no real RSA key needed; HTTP mocked via respx)

connectors/README.md updated to reflect google_sheet is implemented.
---
 connectors/README.md                          |   5 +-
 connectors/google_sheet/README.md             |  75 ++++++
 connectors/google_sheet/__init__.py           |   5 +
 connectors/google_sheet/config.example.yaml   |  22 ++
 connectors/google_sheet/connector.py          | 226 ++++++++++++++++++
 connectors/google_sheet/pytest.ini            |   2 +
 connectors/google_sheet/requirements.txt      |   3 +
 connectors/google_sheet/tests/__init__.py     |   0
 .../google_sheet/tests/test_connector.py      | 166 +++++++++++++
 9 files changed, 502 insertions(+), 2 deletions(-)
 create mode 100644 connectors/google_sheet/README.md
 create mode 100644 connectors/google_sheet/__init__.py
 create mode 100644 connectors/google_sheet/config.example.yaml
 create mode 100644 connectors/google_sheet/connector.py
 create mode 100644 connectors/google_sheet/pytest.ini
 create mode 100644 connectors/google_sheet/requirements.txt
 create mode 100644 connectors/google_sheet/tests/__init__.py
 create mode 100644 connectors/google_sheet/tests/test_connector.py

diff --git a/connectors/README.md b/connectors/README.md
index cfab88a..0b30bdc 100644
--- a/connectors/README.md
+++ b/connectors/README.md
@@ -12,7 +12,7 @@ core sạch và không phụ thuộc vào bất kỳ nền tảng cụ thể nà
 ```
 connectors/
 ├── zalo/            # Zalo OA Connector — nhận/gửi tin nhắn qua Zalo Official Account
-├── google_sheet/    # (sắp có) đọc/ghi đơn hàng, khách hàng qua Google Sheet
+├── google_sheet/    # Google Sheet Connector — đọc/ghi đơn hàng, khách hàng
 ├── nhanh/           # (sắp có) tích hợp Nhanh.vn
 └── base/            # (sắp có) tích hợp Base.vn
 ```
@@ -37,5 +37,6 @@ connectors/
 ```bash
 pip install -e sdk/python[connectors,dev]
 pip install -r connectors/zalo/requirements.txt
-pytest connectors/zalo/tests
+pip install -r connectors/google_sheet/requirements.txt
+pytest connectors/zalo/tests connectors/google_sheet/tests
 ```
diff --git a/connectors/google_sheet/README.md b/connectors/google_sheet/README.md
new file mode 100644
index 0000000..15a14d4
--- /dev/null
+++ b/connectors/google_sheet/README.md
@@ -0,0 +1,75 @@
+# Google Sheet Connector
+
+Đọc/ghi dữ liệu đơn hàng (`Order`) và khách hàng (`Customer`) từ Google Sheet —
+phù hợp cho shop nhỏ chưa dùng phần mềm quản lý bán hàng riêng.
+
+## 1. Chuẩn bị
+
+1. Tạo project trên [Google Cloud Console](https://console.cloud.google.com), bật **Google Sheets API**.
+2. Tạo **Service Account** → tải file JSON key.
+3. Mở Google Sheet cần dùng → **Share** với email của Service Account
+   (dạng `xxx@project-id.iam.gserviceaccount.com`), quyền **Editor**.
+4. Copy `config.example.yaml` → `config.yaml`, điền `spreadsheet_id` (lấy từ URL sheet)
+   và nội dung JSON key.
+
+## 2. Cài đặt
+
+```bash
+pip install -e ../../sdk/python[connectors]
+pip install -r requirements.txt
+```
+
+## 3. Sử dụng
+
+Sheet cần có dòng header ở hàng đầu tiên khớp với tên field, ví dụ tab `Customers`:
+
+| id | name | phone | email | address |
+|----|------|-------|-------|---------|
+| C001 | Nguyễn Văn A | 0901234567 | a@example.com | 12 Lê Lợi, Q1 |
+
+```python
+import asyncio
+import yaml
+from google_sheet.connector import GoogleSheetConnector
+
+async def main():
+    config = yaml.safe_load(open("config.yaml"))
+    async with GoogleSheetConnector(config) as conn:
+        # Đọc toàn bộ khách hàng
+        data = await conn.execute("read_range", {"range": "Customers!A1:E100"})
+        header, *rows = data.get("values", [])
+        customers = [conn.map_to_common_schema({"header": header, "row": r}, "customer") for r in rows]
+        print(customers)
+
+        # Thêm khách hàng mới
+        row = conn.map_from_common_schema(
+            {"id": "C002", "name": "Trần Thị B", "phone": "0909999999"}, "customer"
+        )
+        await conn.execute("append_row", {"range": "Customers!A1:E1", "values": [row]})
+
+asyncio.run(main())
+```
+
+## 4. Các operation hỗ trợ (`execute(operation, params)`)
+
+| operation | params | Mô tả |
+|-----------|--------|-------|
+| `read_range` | `range` (A1 notation, vd `"Orders!A1:H100"`) | Đọc dữ liệu thô |
+| `update_range` | `range`, `values` (list of rows) | Ghi đè một vùng |
+| `append_row` | `range`, `values` | Thêm dòng mới vào cuối bảng |
+| `batch_update` | `updates: [{range, values}, ...]` | Ghi nhiều vùng trong 1 lần gọi API |
+
+## 5. Giới hạn hiện tại
+
+- Mapping Common Schema dựa vào tên cột trong header — đổi tên cột sẽ làm mapping sai,
+  chưa hỗ trợ mapping tùy biến qua config.
+- `Order.items` lưu dưới dạng JSON string trong cột `items_json` (Sheet không có kiểu
+  dữ liệu lồng nhau).
+- Google Sheets API có giới hạn quota (mặc định 60 request ghi/phút/user) —
+  connector sẽ raise `RateLimitError` khi bị giới hạn, cần tự xử lý backoff ở tầng gọi.
+
+## 6. Chạy test
+
+```bash
+pytest tests/
+```
diff --git a/connectors/google_sheet/__init__.py b/connectors/google_sheet/__init__.py
new file mode 100644
index 0000000..73f9ffc
--- /dev/null
+++ b/connectors/google_sheet/__init__.py
@@ -0,0 +1,5 @@
+"""Google Sheet connector for MagiC."""
+
+from .connector import GoogleSheetConnector
+
+__all__ = ["GoogleSheetConnector"]
diff --git a/connectors/google_sheet/config.example.yaml b/connectors/google_sheet/config.example.yaml
new file mode 100644
index 0000000..149223b
--- /dev/null
+++ b/connectors/google_sheet/config.example.yaml
@@ -0,0 +1,22 @@
+# Copy to config.yaml (gitignored) and fill in real values.
+#
+# 1. Tạo Service Account trên Google Cloud Console, bật Google Sheets API.
+# 2. Tải file JSON key của Service Account.
+# 3. Share Google Sheet với email của Service Account (vd:
+#    xxx@my-project.iam.gserviceaccount.com), quyền Editor.
+# 4. Copy nội dung file JSON key vào service_account_info bên dưới, hoặc
+#    dùng service_account_file trỏ tới đường dẫn file JSON.
+
+spreadsheet_id: "YOUR_SPREADSHEET_ID"  # lấy từ URL: /spreadsheets/d/<ID>/edit
+
+# Cách 1: nhúng trực tiếp nội dung JSON key
+service_account_info:
+  type: service_account
+  project_id: "your-project-id"
+  private_key_id: "..."
+  private_key: "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n"
+  client_email: "xxx@your-project.iam.gserviceaccount.com"
+  token_uri: "https://oauth2.googleapis.com/token"
+
+# Cách 2 (thay cho service_account_info): trỏ tới file JSON key
+# service_account_file: "/path/to/service-account-key.json"
diff --git a/connectors/google_sheet/connector.py b/connectors/google_sheet/connector.py
new file mode 100644
index 0000000..e3be059
--- /dev/null
+++ b/connectors/google_sheet/connector.py
@@ -0,0 +1,226 @@
+"""GoogleSheetConnector — read/write orders & customers via Google Sheets API v4.
+
+Auth uses a Google service account (JWT bearer flow via `google-auth`) — the
+sheet must be shared with the service account's email. `google-auth` handles
+token minting/signing (real RSA crypto); this module only makes the actual
+Sheets API calls, over httpx like the other connectors.
+
+Reference: https://developers.google.com/workspace/sheets/api/reference/rest/v4/spreadsheets.values
+- Read:          GET    /v4/spreadsheets/{id}/values/{range}
+- Update:        PUT    /v4/spreadsheets/{id}/values/{range}?valueInputOption=...
+- Append a row:  POST   /v4/spreadsheets/{id}/values/{range}:append?valueInputOption=...
+- Batch update:  POST   /v4/spreadsheets/{id}/values:batchUpdate
+Error body shape is Google's standard `{"error": {"code", "message", "status"}}`.
+"""
+
+import asyncio
+import json
+from typing import Any
+from urllib.parse import quote
+
+import httpx
+from google.auth.transport import Request as GoogleAuthRequestBase
+from google.auth.transport import Response as GoogleAuthResponseBase
+from google.oauth2 import service_account
+
+from magic_ai_sdk.connectors.base import BaseConnector
+from magic_ai_sdk.connectors.errors import AuthError, PermanentError, RateLimitError, TransientError, with_retry
+
+API_BASE = "https://sheets.googleapis.com/v4/spreadsheets"
+SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
+
+
+class _HttpxAuthResponse(GoogleAuthResponseBase):
+    """Adapts an httpx.Response to google.auth.transport.Response."""
+
+    def __init__(self, response: httpx.Response):
+        self._response = response
+
+    @property
+    def status(self) -> int:
+        return self._response.status_code
+
+    @property
+    def headers(self) -> dict[str, str]:
+        return dict(self._response.headers)
+
+    @property
+    def data(self) -> bytes:
+        return self._response.content
+
+
+class _HttpxAuthRequest(GoogleAuthRequestBase):
+    """Adapts httpx (sync client) to the google.auth.transport.Request interface,
+    so credential refresh doesn't need the separate `requests` library — the rest
+    of this connector already standardizes on httpx."""
+
+    def __call__(self, url, method="GET", body=None, headers=None, timeout=None, **kwargs) -> _HttpxAuthResponse:
+        with httpx.Client(timeout=timeout or 15.0) as client:
+            resp = client.request(method, url, content=body, headers=headers)
+        return _HttpxAuthResponse(resp)
+
+# Google's standard error `status` values that mean "back off and retry".
+_RATE_LIMIT_STATUSES = {"RESOURCE_EXHAUSTED"}
+_TRANSIENT_STATUSES = {"UNAVAILABLE", "INTERNAL", "DEADLINE_EXCEEDED", "ABORTED"}
+
+_CUSTOMER_FIELDS = ["id", "name", "phone", "email", "address"]
+_ORDER_FIELDS = ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at"]
+
+
+class GoogleSheetConnector(BaseConnector):
+    name = "google_sheet"
+
+    def __init__(self, config: dict[str, Any]):
+        super().__init__(config)
+        self.spreadsheet_id: str = config["spreadsheet_id"]
+        service_account_info = config.get("service_account_info")
+        service_account_file = config.get("service_account_file")
+        if service_account_info:
+            self._credentials = service_account.Credentials.from_service_account_info(
+                service_account_info, scopes=SCOPES
+            )
+        elif service_account_file:
+            self._credentials = service_account.Credentials.from_service_account_file(
+                service_account_file, scopes=SCOPES
+            )
+        else:
+            raise AuthError("GoogleSheetConnector requires service_account_info or service_account_file")
+        self._client: httpx.AsyncClient | None = None
+        self._token_lock = asyncio.Lock()
+
+    async def connect(self) -> bool:
+        self._client = httpx.AsyncClient(timeout=15.0)
+        await self._ensure_token()
+        return True
+
+    async def disconnect(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
+    async def _ensure_token(self) -> str:
+        async with self._token_lock:
+            if not self._credentials.valid:
+                # google-auth's refresh() is a blocking network call; run off the event loop.
+                await asyncio.to_thread(self._credentials.refresh, _HttpxAuthRequest())
+            return self._credentials.token
+
+    @with_retry(max_attempts=3, base_delay=1.0)
+    async def _request(self, method: str, url: str, **kwargs: Any) -> dict[str, Any]:
+        assert self._client is not None, "call connect() first"
+        token = await self._ensure_token()
+        headers = {"Authorization": f"Bearer {token}"}
+        try:
+            resp = await self._client.request(method, url, headers=headers, **kwargs)
+        except httpx.RequestError as e:
+            raise TransientError(f"network error calling Google Sheets API: {e}") from e
+
+        if resp.status_code == 200:
+            return resp.json() if resp.content else {}
+
+        try:
+            error = resp.json().get("error", {})
+        except (json.JSONDecodeError, ValueError):
+            error = {}
+        status = error.get("status", "")
+        message = error.get("message", f"HTTP {resp.status_code}")
+
+        if resp.status_code == 429 or status in _RATE_LIMIT_STATUSES:
+            raise RateLimitError(f"Google Sheets API rate limited: {message}")
+        if resp.status_code >= 500 or status in _TRANSIENT_STATUSES:
+            raise TransientError(f"Google Sheets API transient error: {message}")
+        raise PermanentError(f"Google Sheets API error ({status or resp.status_code}): {message}")
+
+    def _range_url(self, a1_range: str) -> str:
+        return f"{API_BASE}/{self.spreadsheet_id}/values/{quote(a1_range, safe='')}"
+
+    async def execute(self, operation: str, params: dict[str, Any]) -> Any:
+        if operation == "read_range":
+            return await self._request("GET", self._range_url(params["range"]))
+
+        if operation == "update_range":
+            return await self._request(
+                "PUT",
+                self._range_url(params["range"]),
+                params={"valueInputOption": params.get("value_input_option", "USER_ENTERED")},
+                json={"values": params["values"]},
+            )
+
+        if operation == "append_row":
+            return await self._request(
+                "POST",
+                self._range_url(params["range"]) + ":append",
+                params={"valueInputOption": params.get("value_input_option", "USER_ENTERED")},
+                json={"values": params["values"]},
+            )
+
+        if operation == "batch_update":
+            return await self._request(
+                "POST",
+                f"{API_BASE}/{self.spreadsheet_id}/values:batchUpdate",
+                json={
+                    "valueInputOption": params.get("value_input_option", "USER_ENTERED"),
+                    "data": [{"range": r["range"], "values": r["values"]} for r in params["updates"]],
+                },
+            )
+
+        raise PermanentError(f"unsupported operation: {operation}")
+
+    def map_to_common_schema(self, raw_data: Any, entity_type: str) -> dict[str, Any]:
+        header: list[str] = raw_data["header"]
+        row: list[str] = raw_data["row"]
+        by_col = dict(zip(header, row + [""] * (len(header) - len(row))))
+
+        if entity_type == "customer":
+            known = {f: by_col.pop(f, "") or None for f in _CUSTOMER_FIELDS}
+            return {
+                "id": known["id"] or "",
+                "source_connector": self.name,
+                "external_id": known["id"] or "",
+                "name": known["name"],
+                "phone": known["phone"],
+                "email": known["email"],
+                "address": known["address"],
+                "metadata": by_col,
+            }
+
+        if entity_type == "order":
+            items = []
+            items_raw = by_col.pop("items_json", "")
+            if items_raw:
+                try:
+                    items = json.loads(items_raw)
+                except (json.JSONDecodeError, ValueError):
+                    items = []
+            known = {f: by_col.pop(f, "") for f in _ORDER_FIELDS}
+            return {
+                "id": known["id"] or "",
+                "source_connector": self.name,
+                "external_id": known["id"] or "",
+                "customer_id": known["customer_id"] or None,
+                "status": known["status"] or "pending",
+                "items": items,
+                "total_amount": float(known["total_amount"] or 0),
+                "shipping_fee": float(known["shipping_fee"] or 0),
+                "created_at": known["created_at"] or None,
+                "metadata": by_col,
+            }
+
+        raise PermanentError(f"GoogleSheetConnector cannot map entity_type={entity_type!r}")
+
+    def map_from_common_schema(self, common_data: dict[str, Any], entity_type: str) -> list[str]:
+        if entity_type == "customer":
+            fields = _CUSTOMER_FIELDS
+        elif entity_type == "order":
+            fields = [*_ORDER_FIELDS, "items_json"]
+        else:
+            raise PermanentError(f"GoogleSheetConnector cannot map entity_type={entity_type!r}")
+
+        row = []
+        for field in fields:
+            if field == "items_json":
+                row.append(json.dumps(common_data.get("items", [])))
+            else:
+                value = common_data.get(field, "")
+                row.append("" if value is None else str(value))
+        return row
diff --git a/connectors/google_sheet/pytest.ini b/connectors/google_sheet/pytest.ini
new file mode 100644
index 0000000..2f4c80e
--- /dev/null
+++ b/connectors/google_sheet/pytest.ini
@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode = auto
diff --git a/connectors/google_sheet/requirements.txt b/connectors/google_sheet/requirements.txt
new file mode 100644
index 0000000..bb3ce27
--- /dev/null
+++ b/connectors/google_sheet/requirements.txt
@@ -0,0 +1,3 @@
+# Install the framework first: pip install -e ../../sdk/python[connectors]
+httpx>=0.27
+google-auth>=2.30
diff --git a/connectors/google_sheet/tests/__init__.py b/connectors/google_sheet/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/connectors/google_sheet/tests/test_connector.py b/connectors/google_sheet/tests/test_connector.py
new file mode 100644
index 0000000..bd3f98f
--- /dev/null
+++ b/connectors/google_sheet/tests/test_connector.py
@@ -0,0 +1,166 @@
+"""Unit tests for GoogleSheetConnector — HTTP mocked via respx, auth mocked (no real RSA key)."""
+
+from unittest.mock import MagicMock, patch
+
+import httpx
+import pytest
+import respx
+
+from google_sheet.connector import API_BASE, GoogleSheetConnector
+from magic_ai_sdk.connectors.errors import PermanentError, RateLimitError, TransientError
+
+CONFIG = {"spreadsheet_id": "sheet-123", "service_account_info": {"type": "service_account"}}
+
+
+def _fake_credentials(token="tok-1", valid=True):
+    creds = MagicMock()
+    creds.valid = valid
+    creds.token = token
+
+    def _refresh(request):
+        creds.valid = True
+        creds.token = "refreshed-tok"
+
+    creds.refresh.side_effect = _refresh
+    return creds
+
+
+@pytest.fixture(autouse=True)
+def mock_auth():
+    with patch("google_sheet.connector.service_account.Credentials.from_service_account_info") as m:
+        m.return_value = _fake_credentials()
+        yield m
+
+
+def make_connector(**overrides):
+    return GoogleSheetConnector({**CONFIG, **overrides})
+
+
+async def test_connect_ensures_token():
+    conn = make_connector()
+    async with conn:
+        assert conn._credentials.token == "tok-1"
+
+
+@respx.mock
+async def test_read_range_success():
+    respx.get(f"{API_BASE}/sheet-123/values/Customers%21A1%3AE100").mock(
+        return_value=httpx.Response(200, json={"values": [["id", "name"], ["C1", "Anh"]]})
+    )
+    async with make_connector() as conn:
+        data = await conn.execute("read_range", {"range": "Customers!A1:E100"})
+    assert data["values"][1] == ["C1", "Anh"]
+
+
+@respx.mock
+async def test_append_row_sends_correct_body():
+    route = respx.post(f"{API_BASE}/sheet-123/values/Orders%21A1%3AH1:append").mock(
+        return_value=httpx.Response(200, json={"updates": {"updatedRows": 1}})
+    )
+    async with make_connector() as conn:
+        await conn.execute("append_row", {"range": "Orders!A1:H1", "values": [["O1", "C1", "pending"]]})
+
+    assert route.calls.last.request.url.params["valueInputOption"] == "USER_ENTERED"
+    assert route.calls.last.request.headers["Authorization"] == "Bearer tok-1"
+
+
+@respx.mock
+async def test_batch_update_sends_all_ranges():
+    route = respx.post(f"{API_BASE}/sheet-123/values:batchUpdate").mock(
+        return_value=httpx.Response(200, json={"totalUpdatedRows": 2})
+    )
+    async with make_connector() as conn:
+        await conn.execute(
+            "batch_update",
+            {"updates": [{"range": "A1:A1", "values": [["x"]]}, {"range": "B1:B1", "values": [["y"]]}]},
+        )
+    import json as jsonlib
+
+    body = jsonlib.loads(route.calls.last.request.content)
+    assert len(body["data"]) == 2
+
+
+@respx.mock
+async def test_rate_limit_error():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(
+        return_value=httpx.Response(429, json={"error": {"code": 429, "status": "RESOURCE_EXHAUSTED", "message": "quota"}})
+    )
+    async with make_connector() as conn:
+        with pytest.raises(RateLimitError):
+            await conn.execute("read_range", {"range": "A1"})
+
+
+@respx.mock
+async def test_transient_error_on_unavailable():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(
+        return_value=httpx.Response(503, json={"error": {"code": 503, "status": "UNAVAILABLE", "message": "down"}})
+    )
+    async with make_connector() as conn:
+        with pytest.raises(TransientError):
+            await conn.execute("read_range", {"range": "A1"})
+
+
+@respx.mock
+async def test_permanent_error_on_permission_denied():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(
+        return_value=httpx.Response(403, json={"error": {"code": 403, "status": "PERMISSION_DENIED", "message": "nope"}})
+    )
+    async with make_connector() as conn:
+        with pytest.raises(PermanentError):
+            await conn.execute("read_range", {"range": "A1"})
+
+
+@respx.mock
+async def test_network_error_wrapped_as_transient():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(side_effect=httpx.ConnectError("refused"))
+    async with make_connector() as conn:
+        with pytest.raises(TransientError):
+            await conn.execute("read_range", {"range": "A1"})
+
+
+async def test_expired_credentials_are_refreshed(mock_auth):
+    mock_auth.return_value = _fake_credentials(valid=False)
+    conn = make_connector()
+    async with conn:
+        pass
+    assert conn._credentials.token == "refreshed-tok"
+
+
+def test_map_to_common_schema_customer():
+    conn = make_connector()
+    raw = {"header": ["id", "name", "phone", "email", "address", "note"], "row": ["C1", "Anh", "090", "a@x.com", "HN", "VIP"]}
+    result = conn.map_to_common_schema(raw, "customer")
+    assert result["id"] == "C1"
+    assert result["name"] == "Anh"
+    assert result["metadata"] == {"note": "VIP"}
+
+
+def test_map_to_common_schema_order_with_items():
+    conn = make_connector()
+    raw = {
+        "header": ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at", "items_json"],
+        "row": ["O1", "C1", "confirmed", "100000", "15000", "2026-01-01", '[{"name": "Ao", "quantity": 2}]'],
+    }
+    result = conn.map_to_common_schema(raw, "order")
+    assert result["total_amount"] == 100000.0
+    assert result["items"] == [{"name": "Ao", "quantity": 2}]
+
+
+def test_map_from_common_schema_order_roundtrip():
+    conn = make_connector()
+    row = conn.map_from_common_schema(
+        {"id": "O1", "customer_id": "C1", "status": "pending", "total_amount": 50.0, "items": [{"name": "x"}]},
+        "order",
+    )
+    parsed = conn.map_to_common_schema(
+        {"header": ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at", "items_json"], "row": row},
+        "order",
+    )
+    assert parsed["id"] == "O1"
+    assert parsed["items"] == [{"name": "x"}]
+
+
+def test_map_to_common_schema_rejects_unknown_entity():
+    conn = make_connector()
+    with pytest.raises(PermanentError):
+        conn.map_to_common_schema({"header": [], "row": []}, "case")

From 89d2db5c6a8e142018fbd7df50cc44fa2a15c35a Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 21 Jul 2026 15:15:01 +0000
Subject: [PATCH 2/3] fix(connectors): address review feedback on Google Sheet
 connector
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Fixes the SonarCloud "C Reliability Rating" gate failure and Gemini/
CodeRabbit review comments on PR #18:

- Replace `assert self._client is not None` with an explicit check +
  PermanentError — assertions are stripped under `python -O`, so this
  was flagged as a real reliability bug, not just style
- connect(): close and clear the httpx client if token init fails
  during connect(), instead of leaking it
- _request(): catch GoogleAuthError (covers RefreshError and friends)
  and raise AuthError — a revoked/invalid service account key isn't
  fixed by retrying, so it shouldn't go through with_retry; genuine
  network errors during token refresh still flow through the existing
  httpx.RequestError -> TransientError path since token fetch now runs
  inside the same try block as the request
- _request(): treat HTTP 401 as TransientError and clear the cached
  token so the retry is forced to refresh, instead of failing
  permanently on a token that expired mid-flight
- map_to_common_schema: reject non-dict raw_data, only accept a JSON
  list (not object/other) for items_json, and parse total_amount/
  shipping_fee defensively instead of letting a malformed cell crash
  the whole mapping
- map_from_common_schema: accept Pydantic models (via model_dump/dict)
  in addition to plain dicts, default items to [] instead of "null"
- Simplify redundant `except (json.JSONDecodeError, ValueError)` to
  just ValueError (JSONDecodeError is already a ValueError subclass);
  use resp.is_success instead of `== 200`
- README: warn that config.yaml/service-account key are secrets and
  shouldn't be committed, use repo-root-relative paths consistently,
  and document that map_from_common_schema writes in a fixed column
  order (reordering sheet columns will misalign writes)

9 new/updated tests covering the above; all passing (22 total in this
connector, 36 in sdk/python, 22 in zalo — none affected).
---
 connectors/google_sheet/README.md             | 30 +++++--
 connectors/google_sheet/connector.py          | 73 ++++++++++++----
 .../google_sheet/tests/test_connector.py      | 87 ++++++++++++++++++-
 3 files changed, 166 insertions(+), 24 deletions(-)

diff --git a/connectors/google_sheet/README.md b/connectors/google_sheet/README.md
index 15a14d4..d8b8b40 100644
--- a/connectors/google_sheet/README.md
+++ b/connectors/google_sheet/README.md
@@ -12,11 +12,18 @@ phù hợp cho shop nhỏ chưa dùng phần mềm quản lý bán hàng riêng.
 4. Copy `config.example.yaml` → `config.yaml`, điền `spreadsheet_id` (lấy từ URL sheet)
    và nội dung JSON key.
 
+> ⚠️ **`config.yaml` và file JSON key chứa secret** (private key của service account).
+> Không commit hai file này vào git (đã có trong `.gitignore` mẫu — kiểm tra lại trước khi
+> push). Ở production, ưu tiên `service_account_file` trỏ tới key nằm ngoài repo, hoặc
+> nạp key từ secret manager (Vault, AWS Secrets Manager...) thay vì nhúng thẳng vào YAML.
+
 ## 2. Cài đặt
 
+Chạy từ thư mục gốc repo (`magic/`):
+
 ```bash
-pip install -e ../../sdk/python[connectors]
-pip install -r requirements.txt
+pip install -e 'sdk/python[connectors]'
+pip install -r connectors/google_sheet/requirements.txt
 ```
 
 ## 3. Sử dụng
@@ -27,13 +34,19 @@ Sheet cần có dòng header ở hàng đầu tiên khớp với tên field, ví
 |----|------|-------|-------|---------|
 | C001 | Nguyễn Văn A | 0901234567 | a@example.com | 12 Lê Lợi, Q1 |
 
+> **Quan trọng:** `map_from_common_schema` luôn ghi giá trị theo **thứ tự cột cố định**
+> (đúng thứ tự field trong ví dụ trên với Customer, và
+> `id, customer_id, status, total_amount, shipping_fee, created_at, items_json` với Order).
+> Nếu bạn đổi thứ tự cột trong sheet thật, giá trị sẽ bị ghi lệch cột — giữ đúng thứ tự
+> này, hoặc tự viết lại `map_from_common_schema` để ghi theo header thực tế của sheet.
+
 ```python
 import asyncio
 import yaml
-from google_sheet.connector import GoogleSheetConnector
+from connectors.google_sheet.connector import GoogleSheetConnector
 
 async def main():
-    config = yaml.safe_load(open("config.yaml"))
+    config = yaml.safe_load(open("connectors/google_sheet/config.yaml"))
     async with GoogleSheetConnector(config) as conn:
         # Đọc toàn bộ khách hàng
         data = await conn.execute("read_range", {"range": "Customers!A1:E100"})
@@ -41,7 +54,7 @@ async def main():
         customers = [conn.map_to_common_schema({"header": header, "row": r}, "customer") for r in rows]
         print(customers)
 
-        # Thêm khách hàng mới
+        # Thêm khách hàng mới — thứ tự cột phải khớp header (xem lưu ý ở trên)
         row = conn.map_from_common_schema(
             {"id": "C002", "name": "Trần Thị B", "phone": "0909999999"}, "customer"
         )
@@ -61,8 +74,9 @@ asyncio.run(main())
 
 ## 5. Giới hạn hiện tại
 
-- Mapping Common Schema dựa vào tên cột trong header — đổi tên cột sẽ làm mapping sai,
-  chưa hỗ trợ mapping tùy biến qua config.
+- Mapping Common Schema dựa vào tên cột trong header khi **đọc**; khi **ghi**,
+  `map_from_common_schema` xuất theo thứ tự cột cố định (xem mục 3) — chưa hỗ trợ
+  mapping tùy biến theo header thực tế qua config.
 - `Order.items` lưu dưới dạng JSON string trong cột `items_json` (Sheet không có kiểu
   dữ liệu lồng nhau).
 - Google Sheets API có giới hạn quota (mặc định 60 request ghi/phút/user) —
@@ -71,5 +85,5 @@ asyncio.run(main())
 ## 6. Chạy test
 
 ```bash
-pytest tests/
+pytest connectors/google_sheet/tests
 ```
diff --git a/connectors/google_sheet/connector.py b/connectors/google_sheet/connector.py
index e3be059..1c9c031 100644
--- a/connectors/google_sheet/connector.py
+++ b/connectors/google_sheet/connector.py
@@ -19,6 +19,7 @@
 from urllib.parse import quote
 
 import httpx
+from google.auth.exceptions import GoogleAuthError
 from google.auth.transport import Request as GoogleAuthRequestBase
 from google.auth.transport import Response as GoogleAuthResponseBase
 from google.oauth2 import service_account
@@ -67,6 +68,25 @@ def __call__(self, url, method="GET", body=None, headers=None, timeout=None, **k
 _ORDER_FIELDS = ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at"]
 
 
+def _safe_float(value: Any) -> float:
+    """Malformed sheet cells shouldn't crash the whole mapping — fall back to 0.0."""
+    try:
+        return float(value) if value else 0.0
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def _as_dict(data: Any) -> dict[str, Any]:
+    """Workflows commonly pass a Pydantic Common Schema model here, not a raw dict."""
+    if isinstance(data, dict):
+        return data
+    if hasattr(data, "model_dump"):
+        return data.model_dump()
+    if hasattr(data, "dict"):
+        return data.dict()
+    raise PermanentError("common_data must be a dict or a Pydantic model")
+
+
 class GoogleSheetConnector(BaseConnector):
     name = "google_sheet"
 
@@ -90,7 +110,12 @@ def __init__(self, config: dict[str, Any]):
 
     async def connect(self) -> bool:
         self._client = httpx.AsyncClient(timeout=15.0)
-        await self._ensure_token()
+        try:
+            await self._ensure_token()
+        except Exception:
+            await self._client.aclose()
+            self._client = None
+            raise
         return True
 
     async def disconnect(self) -> None:
@@ -107,24 +132,36 @@ async def _ensure_token(self) -> str:
 
     @with_retry(max_attempts=3, base_delay=1.0)
     async def _request(self, method: str, url: str, **kwargs: Any) -> dict[str, Any]:
-        assert self._client is not None, "call connect() first"
-        token = await self._ensure_token()
-        headers = {"Authorization": f"Bearer {token}"}
+        if self._client is None:
+            raise PermanentError("connector not connected — call connect() first")
+
         try:
+            token = await self._ensure_token()
+            headers = {"Authorization": f"Bearer {token}"}
             resp = await self._client.request(method, url, headers=headers, **kwargs)
         except httpx.RequestError as e:
             raise TransientError(f"network error calling Google Sheets API: {e}") from e
+        except GoogleAuthError as e:
+            # Covers RefreshError and friends — usually a permanently bad/revoked
+            # service account key, not something a retry will fix, so don't retry.
+            raise AuthError(f"Google auth error: {e}") from e
 
-        if resp.status_code == 200:
+        if resp.is_success:
             return resp.json() if resp.content else {}
 
         try:
             error = resp.json().get("error", {})
-        except (json.JSONDecodeError, ValueError):
+        except ValueError:
             error = {}
         status = error.get("status", "")
         message = error.get("message", f"HTTP {resp.status_code}")
 
+        if resp.status_code == 401:
+            # Token expired/revoked mid-flight (credentials.valid was true when we
+            # checked, but Google invalidated it since) — clear it so the retry
+            # forces a fresh refresh instead of reusing the same bad token.
+            self._credentials.token = None
+            raise TransientError(f"Google Sheets API token expired or invalid: {message}")
         if resp.status_code == 429 or status in _RATE_LIMIT_STATUSES:
             raise RateLimitError(f"Google Sheets API rate limited: {message}")
         if resp.status_code >= 500 or status in _TRANSIENT_STATUSES:
@@ -167,8 +204,10 @@ async def execute(self, operation: str, params: dict[str, Any]) -> Any:
         raise PermanentError(f"unsupported operation: {operation}")
 
     def map_to_common_schema(self, raw_data: Any, entity_type: str) -> dict[str, Any]:
-        header: list[str] = raw_data["header"]
-        row: list[str] = raw_data["row"]
+        if not isinstance(raw_data, dict):
+            raise PermanentError("raw_data must be a dict with 'header' and 'row' keys")
+        header: list[str] = raw_data.get("header") or []
+        row: list[str] = raw_data.get("row") or []
         by_col = dict(zip(header, row + [""] * (len(header) - len(row))))
 
         if entity_type == "customer":
@@ -189,9 +228,11 @@ def map_to_common_schema(self, raw_data: Any, entity_type: str) -> dict[str, Any
             items_raw = by_col.pop("items_json", "")
             if items_raw:
                 try:
-                    items = json.loads(items_raw)
-                except (json.JSONDecodeError, ValueError):
-                    items = []
+                    parsed = json.loads(items_raw)
+                except ValueError:
+                    parsed = None
+                if isinstance(parsed, list):
+                    items = parsed
             known = {f: by_col.pop(f, "") for f in _ORDER_FIELDS}
             return {
                 "id": known["id"] or "",
@@ -200,15 +241,17 @@ def map_to_common_schema(self, raw_data: Any, entity_type: str) -> dict[str, Any
                 "customer_id": known["customer_id"] or None,
                 "status": known["status"] or "pending",
                 "items": items,
-                "total_amount": float(known["total_amount"] or 0),
-                "shipping_fee": float(known["shipping_fee"] or 0),
+                "total_amount": _safe_float(known["total_amount"]),
+                "shipping_fee": _safe_float(known["shipping_fee"]),
                 "created_at": known["created_at"] or None,
                 "metadata": by_col,
             }
 
         raise PermanentError(f"GoogleSheetConnector cannot map entity_type={entity_type!r}")
 
-    def map_from_common_schema(self, common_data: dict[str, Any], entity_type: str) -> list[str]:
+    def map_from_common_schema(self, common_data: Any, entity_type: str) -> list[str]:
+        common_data = _as_dict(common_data)
+
         if entity_type == "customer":
             fields = _CUSTOMER_FIELDS
         elif entity_type == "order":
@@ -219,7 +262,7 @@ def map_from_common_schema(self, common_data: dict[str, Any], entity_type: str)
         row = []
         for field in fields:
             if field == "items_json":
-                row.append(json.dumps(common_data.get("items", [])))
+                row.append(json.dumps(common_data.get("items") or []))
             else:
                 value = common_data.get(field, "")
                 row.append("" if value is None else str(value))
diff --git a/connectors/google_sheet/tests/test_connector.py b/connectors/google_sheet/tests/test_connector.py
index bd3f98f..878ef69 100644
--- a/connectors/google_sheet/tests/test_connector.py
+++ b/connectors/google_sheet/tests/test_connector.py
@@ -5,9 +5,10 @@
 import httpx
 import pytest
 import respx
+from google.auth.exceptions import RefreshError
 
 from google_sheet.connector import API_BASE, GoogleSheetConnector
-from magic_ai_sdk.connectors.errors import PermanentError, RateLimitError, TransientError
+from magic_ai_sdk.connectors.errors import AuthError, PermanentError, RateLimitError, TransientError
 
 CONFIG = {"spreadsheet_id": "sheet-123", "service_account_info": {"type": "service_account"}}
 
@@ -164,3 +165,87 @@ def test_map_to_common_schema_rejects_unknown_entity():
     conn = make_connector()
     with pytest.raises(PermanentError):
         conn.map_to_common_schema({"header": [], "row": []}, "case")
+
+
+def test_map_to_common_schema_rejects_non_dict_raw_data():
+    conn = make_connector()
+    with pytest.raises(PermanentError):
+        conn.map_to_common_schema(["not", "a", "dict"], "customer")
+
+
+def test_map_to_common_schema_ignores_non_list_items_json():
+    conn = make_connector()
+    raw = {
+        "header": ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at", "items_json"],
+        "row": ["O1", "C1", "pending", "0", "0", "", '{"not": "a list"}'],
+    }
+    result = conn.map_to_common_schema(raw, "order")
+    assert result["items"] == []
+
+
+def test_map_to_common_schema_malformed_amount_defaults_to_zero():
+    conn = make_connector()
+    raw = {
+        "header": ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at"],
+        "row": ["O1", "C1", "pending", "not-a-number", "also-bad", ""],
+    }
+    result = conn.map_to_common_schema(raw, "order")
+    assert result["total_amount"] == 0.0
+    assert result["shipping_fee"] == 0.0
+
+
+def test_map_from_common_schema_accepts_pydantic_like_model():
+    conn = make_connector()
+
+    class FakeModel:
+        def model_dump(self):
+            return {"id": "C1", "name": "Anh"}
+
+    row = conn.map_from_common_schema(FakeModel(), "customer")
+    assert row[0] == "C1"
+
+
+def test_map_from_common_schema_defaults_none_items_to_empty_list():
+    conn = make_connector()
+    row = conn.map_from_common_schema({"id": "O1", "items": None}, "order")
+    assert row[-1] == "[]"
+
+
+@respx.mock
+async def test_401_response_clears_token_and_raises_transient():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(
+        return_value=httpx.Response(401, json={"error": {"code": 401, "status": "UNAUTHENTICATED", "message": "bad token"}})
+    )
+    async with make_connector() as conn:
+        with pytest.raises(TransientError):
+            await conn.execute("read_range", {"range": "A1"})
+        assert conn._credentials.token is None
+
+
+async def test_connect_failure_closes_client_and_clears_it():
+    with patch("google_sheet.connector.service_account.Credentials.from_service_account_info") as m:
+        creds = _fake_credentials(valid=False)
+        creds.refresh.side_effect = RuntimeError("boom")
+        m.return_value = creds
+
+        conn = GoogleSheetConnector({**CONFIG})
+        with pytest.raises(RuntimeError):
+            await conn.connect()
+        assert conn._client is None
+
+
+async def test_request_without_connect_raises_permanent_error():
+    conn = make_connector()
+    with pytest.raises(PermanentError):
+        await conn.execute("read_range", {"range": "A1"})
+
+
+async def test_refresh_error_during_request_wrapped_as_auth_error():
+    """A revoked/invalid service account key isn't fixed by retrying — surface it
+    as AuthError so the caller stops instead of burning 3 retry attempts."""
+    conn = make_connector()
+    async with conn:
+        conn._credentials.valid = False
+        conn._credentials.refresh.side_effect = RefreshError("revoked")
+        with pytest.raises(AuthError):
+            await conn.execute("read_range", {"range": "A1"})

From 4d4f01517b21c2d917a2dd52734f1c21debeca4e Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 25 Jul 2026 01:47:32 +0000
Subject: [PATCH 3/3] fix(connectors): remove redundant exception classes in
 Zalo connector
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Root cause of the lingering "C Reliability Rating on New Code" gate on
PR #18, which persisted even after the Google Sheet connector's own
redundant excepts were fixed in 89d2db5.

UnicodeDecodeError and json.JSONDecodeError are both subclasses of
ValueError, so listing them alongside ValueError in the same except
tuple is dead weight — SonarCloud rule python:S5713, which it classifies
as a Bug (Reliability), not a code smell. It's the same rule SonarCloud
flagged on PR #17 (connector.py:163) before that PR merged.

Because the Zalo connector landed on main today (PR #17), it still falls
inside SonarCloud's New Code window, so its issues counted against
PR #18's quality gate even though PR #18 doesn't touch those files.

Behavior is unchanged: the same exceptions are still caught, just via
their common base class.

Left sdk/python/magic_ai_sdk/worker.py:131 (same pattern) untouched on
purpose — it's long-standing code outside this PR's scope, and editing
it would pull the whole file into Sonar's new-code window.

Tests: zalo 22, google_sheet 22, sdk/python 36 — all passing.
---
 connectors/zalo/connector.py | 2 +-
 connectors/zalo/webhook.py   | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/connectors/zalo/connector.py b/connectors/zalo/connector.py
index 9126ad8..b96ae2c 100644
--- a/connectors/zalo/connector.py
+++ b/connectors/zalo/connector.py
@@ -189,7 +189,7 @@ def verify_webhook_signature(self, headers: Mapping[str, str], raw_body: bytes)
             if not isinstance(payload, dict):
                 return False
             timestamp = str(payload.get("timestamp", ""))
-        except (UnicodeDecodeError, ValueError):
+        except ValueError:  # UnicodeDecodeError and JSONDecodeError both subclass ValueError
             return False
 
         mac_input = f"{self.app_id}{body_str}{timestamp}{self.oa_secret_key}"
diff --git a/connectors/zalo/webhook.py b/connectors/zalo/webhook.py
index 110579a..cd4ac7a 100644
--- a/connectors/zalo/webhook.py
+++ b/connectors/zalo/webhook.py
@@ -48,7 +48,7 @@ def do_POST(self):
 
             try:
                 payload = json.loads(raw_body)
-            except (json.JSONDecodeError, ValueError):
+            except ValueError:  # json.JSONDecodeError subclasses ValueError
                 self.send_error(400, "Invalid JSON")
                 return