Update code to latestcore data passing model

Author: AndyTitu

example/desktop_app.py

@@ -0,0 +1,32 @@
+# [developer-docs.sdk.python.sdk-import]-start
+from onepassword import *
+import asyncio
+
+
+async def main():
+    # [developer-docs.sdk.python.client-initialization]-start
+    # Connects to the 1Password desktop app.
+    client = await Client.authenticate(
+        auth=DesktopAuth(
+            account_name="AndiTituTest"  # Set to your 1Password account name.
+        ),
+        # Set the following to your own integration name and version.
+        integration_name="My 1Password Integration",
+        integration_version="v1.0.0",
+    )
+    
+    # [developer-docs.sdk.python.list-vaults]-start
+    vaults = await client.vaults.list()
+    for vault in vaults:
+        print(vault)
+    # [developer-docs.sdk.python.list-vaults]-end
+
+    # [developer-docs.sdk.python.list-items]-start
+    overviews = await client.items.list("xw33qlvug6moegr3wkk5zkenoa")
+    for overview in overviews:
+        print(overview.title)
+    # [developer-docs.sdk.python.list-items]-end
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/onepassword/client.py

@@ -25,10 +25,10 @@ async def authenticate(
             integration_version=integration_version,
         )
 
-        if isinstance(auth, str):
+        if isinstance(auth, DesktopAuth):
+            core = DesktopCore(auth.account_name)
+        else:
             core = UniffiCore()
-        elif isinstance(auth, DesktopAuth):
-            core = DesktopCore()
 
         client_id = int(await core.init_client(config))
 

src/onepassword/defaults.py

@@ -34,7 +34,5 @@ def new_default_config(auth: DesktopAuth | str, integration_name, integration_ve
     }
     if isinstance(auth, str):
         client_config_dict["serviceAccountToken"] = auth
-    elif isinstance(auth, DesktopAuth):
-        client_config_dict["accountName"] = auth.account_name
 
     return client_config_dict

src/onepassword/desktop_core.py

@@ -1,32 +1,16 @@
 import ctypes
 import json
 import os
-import platform
-import sys
+import base64
 from ctypes import c_uint8, c_size_t, c_int32, POINTER, byref, c_void_p
 from .core import UniffiCore
+from onepassword.errors import raise_typed_exception
 
 
 def find_1password_lib_path():
-    host_os = platform.system().lower()  # "darwin", "linux", "windows"
-
-    core = UniffiCore()
-    if core is None:
-        raise RuntimeError("failed to get ExtismCore")
-
-    locations_raw = core.invoke({
-        "invocation": {
-            "parameters": {
-                "methodName": "GetDesktopAppIPCClientLocations",
-                "serializedParams": {"host_os": host_os},
-            }
-        }
-    })
-
-    try:
-        locations = json.loads(locations_raw)
-    except Exception as e:
-        raise RuntimeError(f"failed to parse core response: {e}")
+    locations = [
+        "/Users/andititu/core/target/debug/libop_sdk_ipc_client.dylib"
+    ]
 
     for lib_path in locations:
         if os.path.exists(lib_path):
@@ -35,11 +19,12 @@ def find_1password_lib_path():
     raise FileNotFoundError("1Password desktop application not found")
 
 class DesktopCore:
-    def __init__(self):
+    def __init__(self, account_name: str):
         # Determine the path to the desktop app.
         path = find_1password_lib_path()
 
         self.lib = ctypes.CDLL(path)
+        self.account_name = account_name
 
         # Bind the Rust-exported functions
         self.send_message = self.lib.op_sdk_ipc_send_message
@@ -56,43 +41,59 @@ def __init__(self):
         self.free_message.argtypes = [POINTER(c_uint8), c_size_t, c_size_t]
         self.free_message.restype = None
 
-    def call_shared_library(self, payload: bytes) -> bytes:
+    def call_shared_library(self, payload: str, operation_kind: str) -> bytes:
+        # Prepare the input
+        encoded_payload = base64.b64encode(payload.encode("utf-8")).decode("utf-8")
+        data = {
+            "kind": operation_kind,
+            "account_name": self.account_name,
+            "payload": encoded_payload,
+        }
+        message = json.dumps(data).encode("utf-8")
+
+        # Prepare output parameters
         out_buf = POINTER(c_uint8)()
         out_len = c_size_t()
         out_cap = c_size_t()
 
         ret = self.send_message(
-            (ctypes.cast(payload, POINTER(c_uint8))),
-            len(payload),
+            (ctypes.cast(message, POINTER(c_uint8))),
+            len(message),
             byref(out_buf),
             byref(out_len),
             byref(out_cap),
         )
 
         if ret != 0:
-            raise RuntimeError(f"send_message failed with code {ret}")
+            raise RuntimeError(f"send_message failed with code {ret}. Please make sure the Desktop app intehration setting is enabled, or contact 1Password support.")
 
         # Copy bytes into Python
         data = ctypes.string_at(out_buf, out_len.value)
 
         # Free memory via Rust's exported function
         self.free_message(out_buf, out_len, out_cap)
 
-        return data
+        parsed = json.loads(data)
+        payload = bytes(parsed.get("payload", [])).decode("utf-8")
+
+        success = parsed.get("success", False)
+        if not success:
+            raise_typed_exception(Exception(str(payload)))
+
+        return payload
 
-    def init_client(self, config: dict) -> int:
-        payload = json.dumps(config).encode("utf-8")
-        resp = self.call_shared_library(payload)
+    async def init_client(self, config: dict) -> int:
+        payload = json.dumps(config)
+        resp = self.call_shared_library(payload, "init_client")
         return json.loads(resp)
 
-    def invoke(self, invoke_config: dict) -> str:
-        payload = json.dumps(invoke_config).encode("utf-8")
-        resp = self.call_shared_library(payload)
-        return resp.decode("utf-8")
+    async def invoke(self, invoke_config: dict) -> str:
+        payload = json.dumps(invoke_config)
+        return self.call_shared_library(payload, "invoke")
 
     def release_client(self, client_id: int):
-        payload = json.dumps(client_id).encode("utf-8")
+        payload = json.dumps(client_id)
         try:
-            self.call_shared_library(payload)
+            self.call_shared_library(payload, "release_client")
         except Exception as e:
             print(f"failed to release client: {e}")

src/onepassword/types.py

@@ -1159,3 +1159,8 @@ class WordListType(str, Enum):
     """
     Three (random) letter "words"
     """
+
+class VaultListParams(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    decrypt_details: Optional[bool] = Field(alias="decryptDetails", default=None)

src/onepassword/vaults.py

@@ -3,7 +3,7 @@
 from .core import Core
 from typing import Optional, List
 from pydantic import TypeAdapter
-from .types import VaultOverview
+from .types import VaultOverview, VaultListParams
 
 
 class Vaults:
@@ -15,15 +15,15 @@ def __init__(self, client_id, core):
         self.client_id = client_id
         self.core = core
 
-    async def list(self) -> List[VaultOverview]:
+    async def list(self, params: Optional[VaultListParams] = None) -> List[VaultOverview]:
         """
         List all vaults
         """
         response = await self.core.invoke(
             {
                 "invocation": {
                     "clientId": self.client_id,
-                    "parameters": {"name": "VaultsList", "parameters": {}},
+                    "parameters": {"name": "VaultsList", "parameters": {"params": params.model_dump(by_alias=True) if params else {}}},
                 }
             }
         )