Returns the directory containing schema.cypher after EXPORT DATABASE. Older Kuzu releases can create an extra directory named after the export path's basename, e.g. EXPORT DATABASE '/tmp/foo' writes '/tmp/foo/foo/schema.cypher'.
(export_file: str)
| 152 | |
| 153 | |
| 154 | def resolve_export_dir(export_file: str) -> str: |
| 155 | """ |
| 156 | Returns the directory containing schema.cypher after EXPORT DATABASE. |
| 157 | |
| 158 | Older Kuzu releases can create an extra directory named after the export path's |
| 159 | basename, e.g. EXPORT DATABASE '/tmp/foo' writes '/tmp/foo/foo/schema.cypher'. |
| 160 | """ |
| 161 | candidates = [ |
| 162 | export_file, |
| 163 | os.path.join(export_file, os.path.basename(export_file.rstrip(os.sep))), |
| 164 | ] |
| 165 | for candidate in candidates: |
| 166 | schema_file = os.path.join(candidate, "schema.cypher") |
| 167 | if os.path.exists(schema_file): |
| 168 | return candidate |
| 169 | search_root = os.path.dirname(export_file) |
| 170 | for root, dirs, files in os.walk(search_root): |
| 171 | if ".lbug_envs" in dirs: |
| 172 | dirs.remove(".lbug_envs") |
| 173 | if "schema.cypher" in files: |
| 174 | return root |
| 175 | raise ValueError( |
| 176 | "Schema file not found in exported database directories: " |
| 177 | + ", ".join(os.path.join(candidate, "schema.cypher") for candidate in candidates) |
| 178 | ) |
| 179 | |
| 180 | |
| 181 | def get_migration_temp_dir() -> str | None: |
no test coverage detected