dpmcore.loaders.migration

Migration service for importing Access databases into any SQLAlchemy-supported database.

Note

MigrationService lives in dpmcore.loaders (it mutates the database), not in dpmcore.services (which are read-only). It is still re-exported from dpmcore.services for backward compatibility, but the canonical import is from dpmcore.loaders.migration import MigrationService.

MigrationService

class dpmcore.loaders.migration.MigrationService(engine, schema=None)

Bases: object

Migrate an Access database into a SQLAlchemy-managed database.

Unlike other dpmcore services that accept a Session, this service requires an Engine because it needs to run Base.metadata.create_all and DataFrame.to_sql.

Parameters:

• engine (Engine) – A SQLAlchemy Engine.

• schema (str | None)

__init__(engine, schema=None)

Initialise with a SQLAlchemy Engine.

Parameters:

• engine (Engine)

• schema (str | None)

Return type:

None

migrate_from_access(access_path, *, output_path=None)

Extract tables from access_path and load into the database.

Parameters:

• access_path (str) – Filesystem path to an .accdb or .mdb file.

• output_path (Optional[Path]) – Optional final path for the resulting SQLite file. When omitted, the file is renamed to <stem>_<release>_<YYYYMMDD>.db next to the original location. Ignored for non-SQLite engines.

Return type:

MigrationResult

Returns:

A MigrationResult with details of what was loaded.

Raises:

MigrationError – If neither mdb-tools nor pyodbc can read the file.

migrate_from_csv_dir(csv_dir, *, output_path=None)

Load every CSV file from csv_dir into the target database.

Parameters:

• csv_dir (str) – Directory containing one CSV file per table.

• output_path (Optional[Path]) – Optional final path for the resulting SQLite file. See migrate_from_access() for details.

Return type:

MigrationResult

MigrationResult

class dpmcore.loaders.migration.MigrationResult(tables_migrated, total_rows, table_details, warnings, backend_used, database_path=None)

Outcome of a successful migration run.

Parameters:

• tables_migrated (int)

• total_rows (int)

• table_details (Dict[str, int])

• warnings (List[str])

• backend_used (str)

• database_path (Path | None)

tables_migrated: int

total_rows: int

table_details: Dict[str, int]

warnings: List[str]

backend_used: str

database_path: Optional[Path] = None

MigrationError

exception dpmcore.loaders.migration.MigrationError

Bases: Exception

Raised when migration cannot proceed.