(
self,
exp_names: Collection[str],
*entries: Iterable[Union[QueueEntry, QueueDoneResult]],
)
| 583 | return results |
| 584 | |
| 585 | def match_queue_entry_by_name( |
| 586 | self, |
| 587 | exp_names: Collection[str], |
| 588 | *entries: Iterable[Union[QueueEntry, QueueDoneResult]], |
| 589 | ) -> dict[str, Optional[QueueEntry]]: |
| 590 | from funcy import concat |
| 591 | |
| 592 | entry_name_dict: dict[str, QueueEntry] = {} |
| 593 | entry_rev_dict: dict[str, QueueEntry] = {} |
| 594 | for entry in concat(*entries): |
| 595 | if isinstance(entry, QueueDoneResult): |
| 596 | queue_entry: QueueEntry = entry.entry |
| 597 | if entry.result is not None and entry.result.ref_info is not None: |
| 598 | name: Optional[str] = entry.result.ref_info.name |
| 599 | else: |
| 600 | name = queue_entry.name |
| 601 | else: |
| 602 | queue_entry = entry |
| 603 | name = queue_entry.name |
| 604 | if name: |
| 605 | entry_name_dict[name] = queue_entry |
| 606 | entry_rev_dict[queue_entry.stash_rev] = queue_entry |
| 607 | |
| 608 | result: dict[str, Optional[QueueEntry]] = {} |
| 609 | for exp_name in exp_names: |
| 610 | result[exp_name] = None |
| 611 | if exp_name in entry_name_dict: |
| 612 | result[exp_name] = entry_name_dict[exp_name] |
| 613 | continue |
| 614 | if self.scm.is_sha(exp_name): |
| 615 | for rev, entry in entry_rev_dict.items(): |
| 616 | if rev.startswith(exp_name.lower()): |
| 617 | result[exp_name] = entry |
| 618 | break |
| 619 | |
| 620 | return result |
| 621 | |
| 622 | def stash_failed(self, entry: QueueEntry) -> None: |
| 623 | """Add an entry to the failed exp stash. |
no test coverage detected