(idents_file)
| 443 | |
| 444 | |
| 445 | def reap_dbs(idents_file): |
| 446 | log.info(class="st">"Reaping databases...") |
| 447 | |
| 448 | urls = collections.defaultdict(set) |
| 449 | idents = collections.defaultdict(set) |
| 450 | dialects = {} |
| 451 | |
| 452 | with open(idents_file) as file_: |
| 453 | for line in file_: |
| 454 | line = line.strip() |
| 455 | db_name, db_url = line.split(class="st">" ") |
| 456 | url_obj = sa_url.make_url(db_url) |
| 457 | if db_name not in dialects: |
| 458 | dialects[db_name] = url_obj.get_dialect() |
| 459 | dialects[db_name].load_provisioning() |
| 460 | url_key = (url_obj.get_backend_name(), url_obj.host) |
| 461 | urls[url_key].add(db_url) |
| 462 | idents[url_key].add(db_name) |
| 463 | |
| 464 | for url_key in urls: |
| 465 | url = list(urls[url_key])[0] |
| 466 | ident = idents[url_key] |
| 467 | run_reap_dbs(url, ident) |
| 468 | |
| 469 | |
| 470 | @register.init |
nothing calls this directly
no test coverage detected