(
env: Environment,
config: str,
workdir: str,
beanfile: str,
remove_dangling: bool,
workers: int,
verbose: bool,
detailed_report: bool,
)
| 120 | ) |
| 121 | @pass_env |
| 122 | def main( |
| 123 | env: Environment, |
| 124 | config: str, |
| 125 | workdir: str, |
| 126 | beanfile: str, |
| 127 | remove_dangling: bool, |
| 128 | workers: int, |
| 129 | verbose: bool, |
| 130 | detailed_report: bool, |
| 131 | ): |
| 132 | start_time = time.perf_counter() |
| 133 | config_path = pathlib.Path(config) |
| 134 | with config_path.open("rt") as fo: |
| 135 | doc_payload = yaml.safe_load(fo) |
| 136 | import_doc = ImportDoc.model_validate(doc_payload) |
| 137 | workdir_path = pathlib.Path(workdir) |
| 138 | env.logger.info( |
| 139 | "Loaded import doc from [green]%s[/]", |
| 140 | config, |
| 141 | extra={"markup": True, "highlighter": None}, |
| 142 | ) |
| 143 | |
| 144 | generated_txns: list[GeneratedTransaction] = [] |
| 145 | deleted_txns: list[DeletedTransaction] = [] |
| 146 | unprocessed_txns: list[UnprocessedTransaction] = [] |
| 147 | |
| 148 | def handle_import_txn( |
| 149 | txn: GeneratedTransaction | DeletedTransaction | UnprocessedTransaction, |
| 150 | ) -> None: |
| 151 | if isinstance(txn, GeneratedTransaction): |
| 152 | if verbose: |
| 153 | generated_file_path = (workdir_path / txn.file).resolve() |
| 154 | env.logger.info( |
| 155 | "Generated transaction [green]%s[/] to file [green]%s[/]", |
| 156 | txn.id, |
| 157 | strip_base_path(workdir_path.resolve(), generated_file_path), |
| 158 | extra={"markup": True, "highlighter": None}, |
| 159 | ) |
| 160 | generated_txns.append(txn) |
| 161 | elif isinstance(txn, DeletedTransaction): |
| 162 | if verbose: |
| 163 | env.logger.info( |
| 164 | "Deleted transaction [green]%s[/]", |
| 165 | txn.id, |
| 166 | extra={"markup": True, "highlighter": None}, |
| 167 | ) |
| 168 | deleted_txns.append(txn) |
| 169 | elif isinstance(txn, UnprocessedTransaction): |
| 170 | if verbose: |
| 171 | env.logger.info( |
| 172 | "Skipped input transaction %s at [green]%s[/]:[blue]%s[/]", |
| 173 | txn.import_id, |
| 174 | txn.txn.file, |
| 175 | txn.txn.lineno, |
| 176 | extra={"markup": True, "highlighter": None}, |
| 177 | ) |
| 178 | unprocessed_txns.append(txn) |
| 179 | else: |
nothing calls this directly
no test coverage detected