start implement sync client

Shubham · Shubham · commit a9edd550e0c2 · 2025-12-22T19:57:02.000+05:30
diff --git a/objectbox/c.py b/objectbox/c.py
@@ -251,6 +251,123 @@ class OBX_query(ctypes.Structure):
 
 OBX_query_p = ctypes.POINTER(OBX_query)
 
+
+# Sync types
+class OBX_sync(ctypes.Structure):
+    pass
+
+
+OBX_sync_p = ctypes.POINTER(OBX_sync)
+
+
+class OBX_sync_server(ctypes.Structure):
+    pass
+
+
+OBX_sync_server_p = ctypes.POINTER(OBX_sync_server)
+
+
+class OBXSyncCredentialsType(IntEnum):
+    NONE = 1
+    SHARED_SECRET = 2  # Deprecated, use SHARED_SECRET_SIPPED instead
+    GOOGLE_AUTH = 3
+    SHARED_SECRET_SIPPED = 4  # Uses shared secret to create a hashed credential
+    OBX_ADMIN_USER = 5  # ObjectBox admin users (username/password)
+    USER_PASSWORD = 6  # Generic credential type for admin users
+    JWT_ID = 7  # JSON Web Token (JWT): ID token with user identity
+    JWT_ACCESS = 8  # JSON Web Token (JWT): access token for resources
+    JWT_REFRESH = 9  # JSON Web Token (JWT): refresh token
+    JWT_CUSTOM = 10  # JSON Web Token (JWT): custom token type
+
+
+class OBXRequestUpdatesMode(IntEnum):
+    MANUAL = 0  # No updates by default, must call obx_sync_updates_request() manually
+    AUTO = 1  # Same as calling obx_sync_updates_request(sync, TRUE)
+    AUTO_NO_PUSHES = 2  # Same as calling obx_sync_updates_request(sync, FALSE)
+
+
+class OBXSyncState(IntEnum):
+    CREATED = 1
+    STARTED = 2
+    CONNECTED = 3
+    LOGGED_IN = 4
+    DISCONNECTED = 5
+    STOPPED = 6
+    DEAD = 7
+
+
+class OBXSyncCode(IntEnum):
+    OK = 20
+    REQ_REJECTED = 40
+    CREDENTIALS_REJECTED = 43
+    UNKNOWN = 50
+    AUTH_UNREACHABLE = 53
+    BAD_VERSION = 55
+    CLIENT_ID_TAKEN = 61
+    TX_VIOLATED_UNIQUE = 71
+
+
+class OBXSyncError(IntEnum):
+    REJECT_TX_NO_PERMISSION = 1  # Sync client received rejection of transaction writes due to missing permissions
+
+
+class OBXSyncObjectType(IntEnum):
+    FlatBuffers = 1
+    String = 2
+    Raw = 3
+
+
+class OBX_sync_change(ctypes.Structure):
+    _fields_ = [
+        ('entity_id', obx_schema_id),
+        ('puts', ctypes.POINTER(OBX_id_array)),
+        ('removals', ctypes.POINTER(OBX_id_array)),
+    ]
+
+
+class OBX_sync_change_array(ctypes.Structure):
+    _fields_ = [
+        ('list', ctypes.POINTER(OBX_sync_change)),
+        ('count', ctypes.c_size_t),
+    ]
+
+
+class OBX_sync_object(ctypes.Structure):
+    _fields_ = [
+        ('type', ctypes.c_int),  # OBXSyncObjectType
+        ('id', ctypes.c_uint64),
+        ('data', ctypes.c_void_p),
+        ('size', ctypes.c_size_t),
+    ]
+
+
+class OBX_sync_msg_objects(ctypes.Structure):
+    _fields_ = [
+        ('topic', ctypes.c_void_p),
+        ('topic_size', ctypes.c_size_t),
+        ('objects', ctypes.POINTER(OBX_sync_object)),
+        ('count', ctypes.c_size_t),
+    ]
+
+
+class OBX_sync_msg_objects_builder(ctypes.Structure):
+    pass
+
+
+OBX_sync_msg_objects_builder_p = ctypes.POINTER(OBX_sync_msg_objects_builder)
+
+# Define callback types for sync listeners
+OBX_sync_listener_connect = ctypes.CFUNCTYPE(None, ctypes.c_void_p)
+OBX_sync_listener_disconnect = ctypes.CFUNCTYPE(None, ctypes.c_void_p)
+OBX_sync_listener_login = ctypes.CFUNCTYPE(None, ctypes.c_void_p)
+OBX_sync_listener_login_failure = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int)  # arg, OBXSyncCode
+OBX_sync_listener_complete = ctypes.CFUNCTYPE(None, ctypes.c_void_p)
+OBX_sync_listener_error = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int)  # arg, OBXSyncError
+OBX_sync_listener_change = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.POINTER(OBX_sync_change_array))
+OBX_sync_listener_server_time = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int64)
+OBX_sync_listener_msg_objects = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.POINTER(OBX_sync_msg_objects))
+
+
 # manually configure error methods, we can't use `fn()` defined below yet due to circular dependencies
 C.obx_last_error_message.restype = ctypes.c_char_p
 C.obx_last_error_code.restype = obx_err
@@ -310,6 +427,11 @@ def check_obx_err(code: obx_err, func, args) -> obx_err:
         raise create_db_error(code)
     return code
 
+def check_obx_success(code: obx_err) -> bool:
+    if code == DbErrorCode.OBX_NO_SUCCESS:
+        return False
+    check_obx_err(code, None, None)
+    return True
 
 def check_obx_qb_cond(qb_cond: obx_qb_cond, func, args) -> obx_qb_cond:
     """ Raises an exception if obx_qb_cond is not successful. """
@@ -1068,3 +1190,30 @@ def c_array_pointer(py_list: Union[List[Any], np.ndarray], c_type):
 
 OBXBackupRestoreFlags_None = 0
 OBXBackupRestoreFlags_OverwriteExistingData = 1
+
+obx_sync = c_fn("obx_sync", obx_err, [OBX_store_p, ctypes.c_char_p])
+obx_sync_urls = c_fn("obx_sync_urls", obx_err, [OBX_store_p, ctypes.POINTER(ctypes.c_char_p), ctypes.c_size_t])
+
+
+obx_sync_credentials = c_fn_rc('obx_sync_credentials',
+                                [OBX_sync_p, OBXSyncCredentialsType, ctypes.c_void_p, ctypes.c_size_t])
+obx_sync_credentials_user_password = c_fn_rc('obx_sync_credentials_user_password',
+                                               [OBX_sync_p, OBXSyncCredentialsType, ctypes.c_char_p,
+                                                ctypes.c_char_p])
+obx_sync_credentials_add = c_fn_rc('obx_sync_credentials_add',
+                                    [OBX_sync_p, OBXSyncCredentialsType, ctypes.c_void_p, ctypes.c_size_t, ctypes.c_bool])
+obx_sync_credentials_add_user_password = c_fn_rc('obx_sync_credentials_add_user_password',
+                                                   [OBX_sync_p, OBXSyncCredentialsType, ctypes.c_char_p, ctypes.c_char_p,
+                                                    ctypes.c_bool])
+
+obx_sync_request_updates_mode = c_fn_rc('obx_sync_request_updates_mode', [OBX_sync_p, OBXRequestUpdatesMode])
+
+obx_sync_start = c_fn_rc('obx_sync_start', [OBX_sync_p])
+obx_sync_stop = c_fn_rc('obx_sync_stop', [OBX_sync_p])
+
+obx_sync_trigger_reconnect = c_fn_rc('obx_sync_trigger_reconnect', [OBX_sync_p])
+
+obx_sync_protocol_version = c_fn('obx_sync_protocol_version', ctypes.c_uint32, [])
+obx_sync_protocol_version_server = c_fn('obx_sync_protocol_version_server', ctypes.c_uint32, [OBX_sync_p])
+
+obx_sync_close = c_fn_rc('obx_sync_close', [OBX_sync_p])
diff --git a/objectbox/store.py b/objectbox/store.py
@@ -285,3 +285,6 @@ def remove_db_files(db_dir: str):
             Path to DB directory.
         """
         c.obx_remove_db_files(c.c_str(db_dir))
+
+    def c_store(self):
+        return self._c_store
diff --git a/objectbox/sync.py b/objectbox/sync.py
@@ -0,0 +1,163 @@
+import ctypes
+import c as c
+from objectbox import Store
+from objectbox.c import c_array_pointer
+
+
+class SyncCredentials:
+
+    def __init__(self, credential_type: c.OBXSyncCredentialsType):
+        self.type = credential_type
+
+    @staticmethod
+    def none() -> 'SyncCredentials':
+        return SyncCredentialsNone()
+
+    @staticmethod
+    def shared_secret_string(secret: str) -> 'SyncCredentials':
+        return SyncCredentialsSecret(c.OBXSyncCredentialsType.SHARED_SECRET_SIPPED, secret.encode('utf-8'))
+
+    @staticmethod
+    def google_auth(secret: str) -> 'SyncCredentials':
+        return SyncCredentialsSecret(c.OBXSyncCredentialsType.GOOGLE_AUTH, secret.encode('utf-8'))
+
+    @staticmethod
+    def user_and_password(username: str, password: str) -> 'SyncCredentials':
+        return SyncCredentialsUserPassword(c.OBXSyncCredentialsType.USER_PASSWORD, username, password)
+
+    @staticmethod
+    def jwt_id_token(jwt_id_token: str) -> 'SyncCredentials':
+        return SyncCredentialsSecret(c.OBXSyncCredentialsType.JWT_ID, jwt_id_token.encode('utf-8'))
+
+    @staticmethod
+    def jwt_access_token(jwt_access_token: str) -> 'SyncCredentials':
+        return SyncCredentialsSecret(c.OBXSyncCredentialsType.JWT_ACCESS, jwt_access_token.encode('utf-8'))
+
+    @staticmethod
+    def jwt_refresh_token(jwt_refresh_token: str) -> 'SyncCredentials':
+        return SyncCredentialsSecret(c.OBXSyncCredentialsType.JWT_REFRESH, jwt_refresh_token.encode('utf-8'))
+
+    @staticmethod
+    def jwt_custom_token(jwt_custom_token: str) -> 'SyncCredentials':
+        return SyncCredentialsSecret(c.OBXSyncCredentialsType.JWT_CUSTOM, jwt_custom_token.encode('utf-8'))
+
+
+class SyncCredentialsNone(SyncCredentials):
+    def __init__(self):
+        super().__init__(c.OBXSyncCredentialsType.NONE)
+
+
+class SyncCredentialsSecret(SyncCredentials):
+    def __init__(self, credential_type: c.OBXSyncCredentialsType, secret: bytes):
+        super().__init__(credential_type)
+        self.secret = secret
+
+
+class SyncCredentialsUserPassword(SyncCredentials):
+    def __init__(self, credential_type: c.OBXSyncCredentialsType, username: str, password: str):
+        super().__init__(credential_type)
+        self.username = username
+        self.password = password
+
+
+class SyncState:
+    UNKNOWN = 'unknown'
+    CREATED = 'created'
+    STARTED = 'started'
+    CONNECTED = 'connected'
+    LOGGED_IN = 'logged_in'
+    DISCONNECTED = 'disconnected'
+    STOPPED = 'stopped'
+    DEAD = 'dead'
+
+
+class SyncRequestUpdatesMode:
+    MANUAL = 'manual'
+    AUTO = 'auto'
+    AUTO_NO_PUSHES = 'auto_no_pushes'
+
+
+class SyncConnectionEvent:
+    CONNECTED = 'connected'
+    DISCONNECTED = 'disconnected'
+
+
+class SyncLoginEvent:
+    LOGGED_IN = 'logged_in'
+    CREDENTIALS_REJECTED = 'credentials_rejected'
+    UNKNOWN_ERROR = 'unknown_error'
+
+
+class SyncChange:
+    def __init__(self, entity_id: int, entity: type, puts: list[int], removals: list[int]):
+        self.entity_id = entity_id
+        self.entity = entity
+        self.puts = puts
+        self.removals = removals
+
+
+class SyncClient:
+
+    def __init__(self, store: Store, server_urls: list[str], credentials: list[SyncCredentials],
+                 filter_variables: dict[str, str] | None = None):
+        if not server_urls:
+            raise ValueError("Provide at least one server URL")
+
+        if not Sync.is_available():
+            raise RuntimeError(
+                'Sync is not available in the loaded ObjectBox runtime library. '
+                'Please visit https://objectbox.io/sync/ for options.')
+
+        self.__store = store
+        self.__server_urls = server_urls
+        self.__credentials = credentials
+
+        self.__c_sync_client_ptr = c.obx_sync_urls(store.c_store(), c_array_pointer(server_urls, ctypes.c_char_p),
+                                                   len(server_urls))
+
+    def set_credentials(self, credentials: SyncCredentials):
+        if isinstance(credentials, SyncCredentialsNone):
+            c.obx_sync_credentials(self.__c_sync_client_ptr, credentials.type, None, 0)
+        elif isinstance(credentials, SyncCredentialsUserPassword):
+            c.obx_sync_credentials_user_password(self.__c_sync_client_ptr,
+                                                 credentials.type,
+                                                 credentials.username.encode('utf-8'),
+                                                 credentials.password.encode('utf-8'))
+        elif isinstance(credentials, SyncCredentialsSecret):
+            c.obx_sync_credentials(self.__c_sync_client_ptr, credentials.type,
+                                   credentials.secret,
+                                   len(credentials.secret))
+
+    def set_request_updates_mode(self, mode: SyncRequestUpdatesMode):
+        if mode == SyncRequestUpdatesMode.MANUAL:
+            c_mode = c.OBXRequestUpdatesMode.MANUAL
+        elif mode == SyncRequestUpdatesMode.AUTO:
+            c_mode = c.OBXRequestUpdatesMode.AUTO
+        elif mode == SyncRequestUpdatesMode.AUTO_NO_PUSHES:
+            c_mode = c.OBXRequestUpdatesMode.AUTO_NO_PUSHES
+        else:
+            raise ValueError(f"Invalid mode: {mode}")
+        c.obx_sync_request_updates_mode(self.__c_sync_client_ptr, c_mode)
+
+    def start(self):
+        c.obx_sync_start(self.__c_sync_client_ptr)
+
+    def stop(self):
+        c.obx_sync_stop(self.__c_sync_client_ptr)
+
+    def trigger_reconnect(self) -> bool:
+        return c.check_obx_success(c.obx_sync_trigger_reconnect(self.__c_sync_client_ptr))
+
+    @staticmethod
+    def protocol_version() -> int:
+        return c.obx_sync_protocol_version()
+
+    def protocol_server_version(self) -> int:
+        return c.obx_sync_protocol_version_server(self.__c_sync_client_ptr)
+
+    def close(self):
+        c.obx_sync_close(self.__c_sync_client_ptr)
+        self.__c_sync_client_ptr = None
+
+    def is_closed(self) -> bool:
+        return self.__c_sync_client_ptr is None
