| 88 | |
| 89 | |
| 90 | class _FileRegistry: |
| 91 | def __init__(self) -> None: |
| 92 | self._name_mapping: dict[str, set[File]] = defaultdict(set) |
| 93 | self._handlers: set[File] = set() |
| 94 | |
| 95 | @property |
| 96 | def filenames(self) -> list[str]: |
| 97 | return list(self._name_mapping) |
| 98 | |
| 99 | @property |
| 100 | def handlers(self) -> set[File]: |
| 101 | # return set(self._handlers) # return a copy |
| 102 | return self._handlers |
| 103 | |
| 104 | def __len__(self) -> int: |
| 105 | return len(self._handlers) |
| 106 | |
| 107 | def __contains__(self, filename: str) -> bool: |
| 108 | return filename in self.filenames |
| 109 | |
| 110 | def add(self, handler: File) -> None: |
| 111 | self._name_mapping[handler.filename].add(handler) |
| 112 | self._handlers.add(handler) |
| 113 | |
| 114 | def remove(self, handler: File) -> None: |
| 115 | filename = handler.filename |
| 116 | self._name_mapping[filename].remove(handler) |
| 117 | # remove enpty keys |
| 118 | if not self._name_mapping[filename]: |
| 119 | del self._name_mapping[filename] |
| 120 | self._handlers.remove(handler) |
| 121 | |
| 122 | def get_handlers_by_name(self, filename: str) -> set[File]: |
| 123 | # return set(self._name_mapping[filename]) # return a copy |
| 124 | return self._name_mapping[filename] |
| 125 | |
| 126 | def close_all(self) -> None: |
| 127 | handlers = list(self._handlers) # make a copy |
| 128 | for fileh in handlers: |
| 129 | msg = f"Closing remaining open file: {fileh.filename}" |
| 130 | warnings.warn(UnclosedFileWarning(msg)) |
| 131 | fileh.close() |
| 132 | |
| 133 | |
| 134 | # Dict of opened files (keys are filenames and values filehandlers) |