(g, ct_type, entity_type, lbl, *props)
| 57 | return c |
| 58 | |
| 59 | def wait_on_constraint(g, ct_type, entity_type, lbl, *props): |
| 60 | props_filter = "properties = [" + ",".join(["'" + p + "'" for p in props]) + "]" |
| 61 | |
| 62 | q = f"""CALL db.constraints() YIELD type, label, status, properties |
| 63 | WHERE type = '{ct_type}' AND label = '{lbl}' AND status = 'UNDER CONSTRUCTION' |
| 64 | AND {props_filter} RETURN count(1)""" |
| 65 | |
| 66 | while True: |
| 67 | result = g.query(q, read_only=True) |
| 68 | if result.result_set[0][0] == 0: |
| 69 | break |
| 70 | time.sleep(0.5) # sleep 500ms |
| 71 | |
| 72 | def create_constraint(g, ct_type, entity_type, lbl, *props, sync=False): |
| 73 | args = ["GRAPH.CONSTRAINT", "CREATE", g.name, ct_type, entity_type, lbl, "PROPERTIES", len(props)] |
no test coverage detected