sij/sijapi
1
1
Fork 0
Code Issues 3 Pull requests Projects Releases Packages Wiki Activity Actions

Auto-update: Sun Jul 28 15:02:36 PDT 2024

Browse source
This commit is contained in:
sanj 2024-07-28 15:02:36 -07:00
parent 5792657c6e
commit b059d93f37
2 changed files with 56 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
sijapi/__main__.py
View file

@ -69,12 +69,13 @@ async def lifespan(app: FastAPI):
# Check if other instances have more recent data # Check if other instances have more recent data
source = await API.get_most_recent_source() source = await API.get_most_recent_source()
if source: if source:
crit(f"Pulling changes from {source['ts_id']}...") crit(f"Pulling changes from {source['ts_id']} ({source['ts_ip']})...")
await API.pull_changes(source) total_changes = await API.pull_changes(source)
crit("Data pull complete.") crit(f"Data pull complete. Total changes: {total_changes}")
else: else:
crit("No instances with more recent data found.") crit("No instances with more recent data found.")
except Exception as e: except Exception as e:
crit(f"Error during startup: {str(e)}") crit(f"Error during startup: {str(e)}")
crit(f"Traceback: {traceback.format_exc()}") crit(f"Traceback: {traceback.format_exc()}")

67
sijapi/classes.py
View file

@ -378,6 +378,17 @@ class APIConfig(BaseModel):
return most_recent_source return most_recent_source
async def pull_changes(self, source_pool_entry): async def pull_changes(self, source_pool_entry):
total_inserts = 0
total_updates = 0
table_changes = {}
source_id = source_pool_entry['ts_id']
source_ip = source_pool_entry['ts_ip']
dest_id = os.environ.get('TS_ID')
dest_ip = self.local_db['ts_ip']
info(f"Starting sync from source {source_id} ({source_ip}) to destination {dest_id} ({dest_ip})")
async with self.get_connection(source_pool_entry) as source_conn: async with self.get_connection(source_pool_entry) as source_conn:
async with self.get_connection() as dest_conn: async with self.get_connection() as dest_conn:
tables = await source_conn.fetch(""" tables = await source_conn.fetch("""
@ -387,36 +398,62 @@ class APIConfig(BaseModel):
for table in tables: for table in tables:
table_name = table['tablename'] table_name = table['tablename']
last_synced_version = await self.get_last_synced_version(table_name, source_pool_entry['ts_id']) inserts = 0
updates = 0
last_synced_version = await self.get_last_synced_version(table_name, source_id)
changes = await source_conn.fetch(f""" changes = await source_conn.fetch(f"""
SELECT * FROM "{table_name}" SELECT * FROM "{table_name}"
WHERE version > $1 AND server_id = $2 WHERE version > $1 AND server_id = $2
ORDER BY version ASC ORDER BY version ASC
""", last_synced_version, source_pool_entry['ts_id']) """, last_synced_version, source_id)
for change in changes: for change in changes:
columns = list(change.keys()) columns = list(change.keys())
values = [change[col] for col in columns] values = [change[col] for col in columns]
# Log the target database and table name debug(f"Syncing data for table: {table_name} from {source_id} to {dest_id}")
debug(f"Attempting to insert data into table: {table_name} in database: {dest_conn._params['database']} (host: {dest_conn._params['host']})")
insert_query = f""" if table_name == 'sync_status':
INSERT INTO "{table_name}" ({', '.join(columns)}) insert_query = f"""
VALUES ({', '.join(f'${i+1}' for i in range(len(columns)))}) INSERT INTO "{table_name}" ({', '.join(columns)})
ON CONFLICT (id) DO UPDATE SET VALUES ({', '.join(f'${i+1}' for i in range(len(columns)))})
{', '.join(f"{col} = EXCLUDED.{col}" for col in columns if col != 'id')} ON CONFLICT (table_name, server_id) DO UPDATE SET
""" {', '.join(f"{col} = EXCLUDED.{col}" for col in columns if col not in ['table_name', 'server_id'])}
"""
else:
insert_query = f"""
INSERT INTO "{table_name}" ({', '.join(columns)})
VALUES ({', '.join(f'${i+1}' for i in range(len(columns)))})
ON CONFLICT (id) DO UPDATE SET
{', '.join(f"{col} = EXCLUDED.{col}" for col in columns if col != 'id')}
"""
try: try:
await dest_conn.execute(insert_query, *values) result = await dest_conn.execute(insert_query, *values)
except asyncpg.exceptions.UndefinedColumnError as e: if 'UPDATE' in result:
err(f"UndefinedColumnError in table: {table_name} in database: {dest_conn._params['database']} (host: {dest_conn._params['host']})") updates += 1
raise e else:
inserts += 1
except Exception as e:
err(f"Error syncing data for table {table_name} from {source_id} to {dest_id}: {str(e)}")
if changes: if changes:
await self.update_sync_status(table_name, source_pool_entry['ts_id'], changes[-1]['version']) await self.update_sync_status(table_name, source_id, changes[-1]['version'])
total_inserts += inserts
total_updates += updates
table_changes[table_name] = {'inserts': inserts, 'updates': updates}
info(f"Synced {table_name} from {source_id} to {dest_id}: {inserts} inserts, {updates} updates")
info(f"Sync complete from {source_id} ({source_ip}) to {dest_id} ({dest_ip})")
info(f"Total changes: {total_inserts} inserts, {total_updates} updates")
info("Changes by table:")
for table, changes in table_changes.items():
info(f" {table}: {changes['inserts']} inserts, {changes['updates']} updates")
return total_inserts + total_updates
async def push_changes_to_all(self): async def push_changes_to_all(self):