Add separate method to indicate if a branch are in sync across workers

changelog/+b21b1e10.added.md

@@ -0,0 +1 @@
+Added InfrahubClient.schema.in_sync method to indicate if a specific branch is in sync across all worker types

infrahub_sdk/schema.py

@@ -637,8 +637,7 @@
         """Wait until the schema has converged on the selected branch or the timeout has been reached"""
         waited = 0
         while True:
-            status = await self.client.execute_graphql(query=SCHEMA_HASH_SYNC_STATUS, branch_name=branch)
-            if status["InfrahubStatus"]["summary"]["schema_hash_synced"]:
+            if await self.in_sync(branch=branch):
                 self.client.log.info(f"Schema successfully converged after {waited} seconds")
                 return
 
@@ -649,6 +648,11 @@
             waited += 1
             await asyncio.sleep(delay=1)
 
+    async def in_sync(self, branch: Optional[str] = None) -> bool:
+        """Indicate if the schema is in sync across all workers for the provided branch"""
+        response = await self.client.execute_graphql(query=SCHEMA_HASH_SYNC_STATUS, branch_name=branch)
+        return response["InfrahubStatus"]["summary"]["schema_hash_synced"]
+
     async def check(self, schemas: list[dict], branch: Optional[str] = None) -> tuple[bool, Optional[dict]]:
         branch = branch or self.client.default_branch
         url = f"{self.client.address}/api/schema/check?branch={branch}"
@@ -1041,8 +1045,7 @@
         """Wait until the schema has converged on the selected branch or the timeout has been reached"""
         waited = 0
         while True:
-            status = self.client.execute_graphql(query=SCHEMA_HASH_SYNC_STATUS, branch_name=branch)
-            if status["InfrahubStatus"]["summary"]["schema_hash_synced"]:
+            if self.in_sync(branch=branch):
                 self.client.log.info(f"Schema successfully converged after {waited} seconds")
                 return
 
@@ -1053,6 +1056,11 @@
             waited += 1
             sleep(1)
 
+    def in_sync(self, branch: Optional[str] = None) -> bool:
+        """Indicate if the schema is in sync across all workers for the provided branch"""
+        response = self.client.execute_graphql(query=SCHEMA_HASH_SYNC_STATUS, branch_name=branch)
+        return response["InfrahubStatus"]["summary"]["schema_hash_synced"]
+
     def check(self, schemas: list[dict], branch: Optional[str] = None) -> tuple[bool, Optional[dict]]:
         branch = branch or self.client.default_branch
         url = f"{self.client.address}/api/schema/check?branch={branch}"