ashatora/xmlrpcserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add logging to replication

Browse Source
This commit is contained in:
Ivan Vazhenin
2023-11-22 20:32:08 +03:00
parent 5363cbc10f
commit 004eeb4b2b
1 changed files with 15 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
main.py
View File

@@ -81,15 +81,16 @@ def run_tasks():
def replication_task(): def replication_task():
while True: while True:
conn = db.connect_db() conn = db.connect_db()
logging.warning(connected)
with Session(conn) as session: with Session(conn) as session:
for bndname in connected: for item in session.query(db.Queue).join(db.Queue.user).all():
for item in session.query(db.Queue).join(db.Queue.user).filter_by(bndname=bndname).all(): bndname = item.user.bndname
if item.user.is_active_now(): if item.user.is_active_now() and bndname != Config.self_bnd:
if item.user.newbnd: if item.user.newbnd:
replication(bndname, item.commit_id, item.schema) replication(bndname, item.commit_id, item.schema)
else: else:
replication_old(bndname, item.commit_id, item.schema) replication_old(bndname, item.commit_id, item.schema)
session.delete(item) session.delete(item)
session.commit() session.commit()
time.sleep(60) time.sleep(60)
@@ -118,7 +119,7 @@ def crc32(filename, chunk_size=65536):
return "%08X" % (checksum & 0xFFFFFFFF) return "%08X" % (checksum & 0xFFFFFFFF)
def send_object_replication(ws, ids): def send_object_replication(bndname, ws, ids):
qu = GroupQuery(Envelope()) qu = GroupQuery(Envelope())
query = GroupQuery(Envelope.world()) query = GroupQuery(Envelope.world())
query.setUids(ids) query.setUids(ids)
@@ -152,12 +153,12 @@ def send_object_replication(ws, ids):
'Name': c1000[0]['fileName'], 'Name': c1000[0]['fileName'],
'Type': 'OOD', 'Type': 'OOD',
'metadata_version': '1', 'metadata_version': '1',
'source': 'bnd127', 'source': Config.self_bnd,
'system_date': str(created_date.timestamp()), 'system_date': str(created_date.timestamp()),
}) })
xml_version = ET.SubElement(xml_objects, 'version', { xml_version = ET.SubElement(xml_objects, 'version', {
'object_id': chart['uid'].replace('-', ''), 'object_id': chart['uid'].replace('-', ''),
'source': 'bnd127', 'source': Config.self_bnd,
'system_date': str(created_date.timestamp()), 'system_date': str(created_date.timestamp()),
'version': '1.0', 'version': '1.0',
'version_id': chart['uid'].replace('-', ''), 'version_id': chart['uid'].replace('-', ''),
@@ -206,7 +207,7 @@ def send_object_replication(ws, ids):
}) })
params = { params = {
'from': f'tcp://{Config.self_bnd}', 'from': f'tcp://{Config.self_bnd}',
'to': f'tcp://{Config.remote_bnd}', 'to': f'tcp://{bndname}',
'ts_added': date.timestamp(), 'ts_added': date.timestamp(),
'user_id': '0', 'user_id': '0',
'query_type': NEW_REPLICATION_REQUEST, 'query_type': NEW_REPLICATION_REQUEST,
@@ -235,12 +236,13 @@ def replication_old(bnd_name: str, commit_id: str, schema: str):
logger.warning(res) logger.warning(res)
created, updated, _ = ws.changes(commit_id) created, updated, _ = ws.changes(commit_id)
ids = list(set(created) | set(updated)) ids = list(set(created) | set(updated))
send_object_replication(ws, ids) send_object_replication(bnd_name, ws, ids)
ws.clearData(True) ws.clearData(True)
logger.warning('Replication to old bnd is sent') logger.warning('Replication to old bnd is sent')
def replication(bnd_name: str, commit_id: str, schema: str): def replication(bnd_name: str, commit_id: str, schema: str):
logging.warning(f'{bnd_name} {commit_id} {schema}')
date = datetime.datetime.now() date = datetime.datetime.now()
rxmls = RequestXmlService() rxmls = RequestXmlService()
res_id = uuid4().hex res_id = uuid4().hex