(self)
| 108 | self.dispatch_workers(file_list) |
| 109 | |
| 110 | def single_copy(self): |
| 111 | while True: |
| 112 | file = self.file_queue.get() |
| 113 | dest_path = self.dest_dir / file.relative_to(self.src_dir) |
| 114 | if self.replace: |
| 115 | dest_path.unlink(missing_ok=True) |
| 116 | if not dest_path.exists(): |
| 117 | try: |
| 118 | shutil.copy2(file, str(dest_path.parent), follow_symlinks=False) |
| 119 | except: |
| 120 | print(f'Couldn\'t copy "{file}"') |
| 121 | print_exc() |
| 122 | if self.delete: |
| 123 | file.unlink() |
| 124 | self.file_queue.task_done() |
| 125 | with self._lock: |
| 126 | self.progress_bar.update(1) |
| 127 | |
| 128 | def dispatch_workers(self, |
| 129 | file_list: List[str]): |
nothing calls this directly
no outgoing calls
no test coverage detected