Add logging to replication
This commit is contained in:
@@ -331,14 +331,18 @@ def replication(bnd_name: str, commit_id: str, schema: str):
|
|||||||
|
|
||||||
def pika_callback(ch, method, properties, body):
|
def pika_callback(ch, method, properties, body):
|
||||||
commit_info = json.loads(body)
|
commit_info = json.loads(body)
|
||||||
|
logging.warning(commit_info)
|
||||||
schema = commit_info.get('schema') or Config.oodb_schema
|
schema = commit_info.get('schema') or Config.oodb_schema
|
||||||
commit = commit_info['commit']
|
commit = commit_info['commit']
|
||||||
conn = db.connect_db()
|
conn = db.connect_db()
|
||||||
with Session(conn) as session:
|
with Session(conn) as session:
|
||||||
for user in session.query(db.User).filter(db.User.active == True, db.User.upstream == False).all():
|
for user in session.query(db.User).filter(db.User.active == True).all():
|
||||||
|
if user.bndname == Config.self_bnd:
|
||||||
|
continue
|
||||||
profiles = {x.scheme: x.to_dict() for x in user.profiles}
|
profiles = {x.scheme: x.to_dict() for x in user.profiles}
|
||||||
if len(profiles) == 0 or schema in profiles:
|
if len(profiles) == 0 or schema in profiles:
|
||||||
item = db.Queue(user_id=user.id, commit_id=commit, schema=schema)
|
item = db.Queue(user_id=user.id, commit_id=commit, schema=schema)
|
||||||
|
logging.warning(item)
|
||||||
session.add(item)
|
session.add(item)
|
||||||
session.commit()
|
session.commit()
|
||||||
ch.basic_ack(delivery_tag=method.delivery_tag)
|
ch.basic_ack(delivery_tag=method.delivery_tag)
|
||||||
|
|||||||
Reference in New Issue
Block a user