(working_dir,entity_set)
| 323 | else: |
| 324 | return ret[0] |
| 325 | def search_chunks(working_dir,entity_set): |
| 326 | db = pymysql.connect(host='localhost',port=4321, user='root', |
| 327 | passwd='123', charset='utf8mb4') |
| 328 | res=[] |
| 329 | db_name=os.path.basename(working_dir) |
| 330 | cursor = db.cursor() |
| 331 | for entity in entity_set: |
| 332 | if entity=='root': |
| 333 | continue |
| 334 | sql=f"select source_id from {db_name}.entities where entity_name=%s " |
| 335 | cursor.execute(sql,(entity,)) |
| 336 | ret=cursor.fetchall() |
| 337 | res.append(ret[0]) |
| 338 | return res |
| 339 | def search_nodes(entity_set,working_dir): |
| 340 | db = pymysql.connect(host='localhost',port=4321, user='root', |
| 341 | passwd='123', charset='utf8mb4') |
nothing calls this directly
no outgoing calls
no test coverage detected