Use profiles for replication

main.py

@@ -62,6 +62,15 @@ def upload_file(filename: str, key: str, bucket: str):
         client.put_object(Body=f.read(), Bucket=bucket, Key=key)
 
 
+def get_branch(bndname: str, scheme: str):
+    conn = db.connect_db()
+    with Session(conn) as session:
+        item = session.query(db.IncomeBranch).filter_by(scheme=scheme).join(db.User).filter_by(bndname=bndname).one_or_none()
+        if item:
+            return item.branch
+        return None
+
+
 def run_tasks():
     logger.debug('Task thread started.')
     while True:
@@ -325,6 +334,8 @@ def pika_callback(ch, method, properties, body):
     conn = db.connect_db()
     with Session(conn) as session:
         for user in session.query(db.User).filter(db.User.active == True, db.User.upstream == False).all():
+            profiles = {x.scheme: x.to_dict() for x in user.profiles}
+            if len(profiles) == 0 or schema in profiles:
                 item = db.Queue(user_id=user.id, commit_id=commit, schema=schema)
                 session.add(item)
         session.commit()
@@ -543,10 +554,8 @@ def put_object(params, files, url):
 
 
 def apply_commits(params, files, url):
-    logger.warning(params)
     assert len(files) == 1
     file = files[0]
-    logger.warning(file)
     dir = TemporaryDirectory()
     r = requests.get(file['url'])
     with zipfile.ZipFile(io.BytesIO(r.content)) as zip_ref:
@@ -555,8 +564,7 @@ def apply_commits(params, files, url):
     repl = req.find('replication')
     scheme = repl.get('scheme')
     commit = repl.get('id')
-    logger.warning(scheme)
-    logger.warning(commit)
+    branch = get_branch(params['from'].replace('tcp://', ''), scheme)
     con = OOConnectionParams(scheme, Config.oodb_host, Config.oodb_port, Config.oodb_dbname,
                              Config.oodb_username, Config.oodb_passwd, scheme)
     ws = OODBWorkspace.ws(scheme)