Load a scenario's data (or the default) into CouchDB. Returns {collection: (db, n)}. Resolves the manifest first, so an unknown ``scenario_id`` raises FileNotFoundError before anything is dropped. ``reset_first=True`` then drops databases so collections absent from the manifest are left
(
scenario_id=None,
force: bool = True,
reset_first: bool = False,
managed_only: bool = False,
)
| 127 | # Load |
| 128 | # --------------------------------------------------------------------------- # |
| 129 | def init_data( |
| 130 | scenario_id=None, |
| 131 | force: bool = True, |
| 132 | reset_first: bool = False, |
| 133 | managed_only: bool = False, |
| 134 | ) -> dict: |
| 135 | """Load a scenario's data (or the default) into CouchDB. Returns {collection: (db, n)}. |
| 136 | |
| 137 | Resolves the manifest first, so an unknown ``scenario_id`` raises FileNotFoundError |
| 138 | before anything is dropped. ``reset_first=True`` then drops databases so collections |
| 139 | absent from the manifest are left empty rather than carrying over. |
| 140 | """ |
| 141 | manifest, base_dir = _resolve_manifest( |
| 142 | scenario_id |
| 143 | ) # validate first (raises on unknown id) |
| 144 | if reset_first: |
| 145 | reset(managed_only=managed_only) |
| 146 | results = {} |
| 147 | for key, spec in manifest.items(): |
| 148 | results[key] = loader.load_collection( |
| 149 | key, spec, drop=force, base_dir=base_dir |
| 150 | ) # database name = key |
| 151 | logger.info( |
| 152 | "Scenario %s: '%s' → %s (%d docs).", scenario_id, key, *results[key] |
| 153 | ) |
| 154 | return results |
| 155 | |
| 156 | |
| 157 | def main() -> None: |
no test coverage detected