47 lines
1.9 KiB
Python
47 lines
1.9 KiB
Python
from engine.types.operation import Operation
|
|
|
|
|
|
class SyncEngine:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def sync(
|
|
self, cursor: int, operations: list[Operation]
|
|
) -> tuple[int, list[Operation]]:
|
|
"""Sync the local database with the server's database.
|
|
|
|
Args:
|
|
cursor (int): The cursor of the last operation that was applied to the local database.
|
|
operations (list[Operation]): The list of operations that were performed on the local database since the last sync.
|
|
|
|
Returns:
|
|
tuple[int, list[Operation]]: A tuple containing the new cursor and the list of operations that need to be applied to the local database.
|
|
"""
|
|
# For now, just return the same cursor and an empty list of operations.
|
|
new_operations = self.get_operations_after_cursor(cursor)
|
|
if len(operations) > 0:
|
|
self.apply_operations(operations)
|
|
return cursor + len(new_operations), new_operations
|
|
|
|
def apply_operations(self, operations: list[Operation]) -> None:
|
|
"""Apply the given operations to the local database.
|
|
|
|
Args:
|
|
operations (list[Operation]): The list of operations to apply to the local database.
|
|
"""
|
|
# For now, just print the operations.
|
|
for operation in operations:
|
|
print("APPLYING OPERATION:", operation.model_dump(), flush=True)
|
|
|
|
def get_operations_after_cursor(self, cursor: int) -> list[Operation]:
|
|
"""Get the list of operations that were performed on the server's database after the given cursor.
|
|
|
|
Args:
|
|
cursor (int): The cursor of the last operation that was applied to the local database.
|
|
|
|
Returns:
|
|
list[Operation]: The list of operations that were performed on the server's database after the given cursor.
|
|
"""
|
|
# For now, just return an empty list of operations.
|
|
return []
|