"""Access-to-database migration service.
Extracts tables from a Microsoft Access ``.accdb`` / ``.mdb`` file and
loads them into any SQLAlchemy-supported database, preserving the ORM
schema created by ``Base.metadata.create_all``.
Two extraction backends are tried in order:
1. **mdb-tools** (``mdb-tables`` + ``mdb-export``) — works on Linux
without ODBC drivers.
2. **pyodbc** — works on Windows / macOS with the Access ODBC driver
installed.
Both ``pandas`` and ``pyodbc`` are optional dependencies; they are
imported lazily so the rest of dpmcore works without them.
"""
from __future__ import annotations
import logging
import re
import shutil
import subprocess
from dataclasses import dataclass
from datetime import date, datetime, timezone
from io import StringIO
from pathlib import Path
from typing import Any, Dict, List, Optional
from sqlalchemy import inspect, select, text
from sqlalchemy.engine import Engine
from dpmcore.orm.base import Base
from dpmcore.orm.infrastructure import Release
logger = logging.getLogger(__name__)
# Access system tables that should never be migrated.
_SYSTEM_TABLE_PREFIXES = ("MSys", "~")
[docs]
class MigrationError(Exception):
"""Raised when migration cannot proceed."""
[docs]
@dataclass(frozen=True)
class MigrationResult:
"""Outcome of a successful migration run."""
tables_migrated: int
total_rows: int
table_details: Dict[str, int]
warnings: List[str]
backend_used: str
database_path: Optional[Path] = None
[docs]
class MigrationService:
"""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``.
Args:
engine: A SQLAlchemy :class:`~sqlalchemy.engine.Engine`.
"""
[docs]
def __init__(self, engine: Engine, schema: str | None = None) -> None:
"""Initialise with a SQLAlchemy Engine."""
self._engine = engine
self._schema = schema
self._ddl_engine = (
engine.execution_options(schema_translate_map={None: schema})
if schema is not None
else engine
)
# -------------------------------------------------------------- #
# Public API
# -------------------------------------------------------------- #
[docs]
def migrate_from_access(
self,
access_path: str,
*,
output_path: Optional[Path] = None,
) -> MigrationResult:
"""Extract tables from *access_path* and load into the database.
Args:
access_path: Filesystem path to an ``.accdb`` or ``.mdb``
file.
output_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.
Returns:
A :class:`MigrationResult` with details of what was loaded.
Raises:
MigrationError: If neither mdb-tools nor pyodbc can read
the file.
"""
data, backend = self._extract_tables(access_path)
return self._finalize(data, backend=backend, output_path=output_path)
[docs]
def migrate_from_csv_dir(
self,
csv_dir: str,
*,
output_path: Optional[Path] = None,
) -> MigrationResult:
"""Load every CSV file from *csv_dir* into the target database.
Args:
csv_dir: Directory containing one CSV file per table.
output_path: Optional final path for the resulting SQLite
file. See :meth:`migrate_from_access` for details.
"""
path = Path(csv_dir)
if not path.exists():
raise MigrationError(f"CSV directory '{csv_dir}' does not exist.")
if not path.is_dir():
raise MigrationError(f"CSV path '{csv_dir}' is not a directory.")
data = self._extract_from_csv_dir(path)
if not data:
raise MigrationError(f"No CSV files found in '{csv_dir}'.")
return self._finalize(
data,
backend="csv",
output_path=output_path,
)
# -------------------------------------------------------------- #
# Extraction
# -------------------------------------------------------------- #
def _extract_tables(self, access_path: str) -> tuple[Dict[str, Any], str]:
"""Try mdb-tools first, fall back to pyodbc."""
try:
data = self._extract_with_mdbtools(access_path)
return data, "mdbtools"
except (FileNotFoundError, OSError, subprocess.CalledProcessError):
logger.debug("mdb-tools not available, falling back to pyodbc")
try:
data = self._extract_with_pyodbc(access_path)
return data, "pyodbc"
except Exception as exc:
raise MigrationError(
"Could not read the Access file. Install mdb-tools "
"(Linux) or the Microsoft Access ODBC driver "
"(Windows/macOS), then try again."
) from exc
def _extract_with_mdbtools(self, access_path: str) -> Dict[str, Any]:
"""Use ``mdb-tables`` / ``mdb-export`` (subprocess)."""
import pandas as pd # lazy
raw = subprocess.check_output( # noqa: S603
["mdb-tables", "-1", access_path], # noqa: S607
text=True,
)
table_names = [
t.strip()
for t in raw.strip().split("\n")
if t.strip()
and not any(
t.strip().startswith(p) for p in _SYSTEM_TABLE_PREFIXES
)
]
data: Dict[str, Any] = {}
for table in table_names:
csv_text = subprocess.check_output( # noqa: S603
["mdb-export", access_path, table], # noqa: S607
text=True,
)
df = pd.read_csv(StringIO(csv_text), dtype=str)
# Attempt numeric conversion where possible.
data[table] = self._coerce_numeric_columns(df)
return data
def _extract_with_pyodbc(self, access_path: str) -> Dict[str, Any]:
"""Use pyodbc with the Access ODBC driver."""
import decimal
import pandas as pd # lazy
import pyodbc # lazy
conn_str = (
r"DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};"
f"DBQ={access_path};"
)
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()
# Discover user tables (skip system tables).
tables = [
row.table_name
for row in cursor.tables(tableType="TABLE")
if not any(
row.table_name.startswith(p) for p in _SYSTEM_TABLE_PREFIXES
)
]
numeric_types = (int, float, decimal.Decimal)
data: Dict[str, Any] = {}
for table in tables:
cursor.execute(f"SELECT * FROM [{table}]") # noqa: S608
col_meta = cursor.description
col_names = [col[0] for col in col_meta]
col_types = [col[1] for col in col_meta]
rows = cursor.fetchall()
df = pd.DataFrame.from_records(rows, columns=col_names)
# Apply schema-based type enforcement: keep text columns
# as text even when values look numeric.
for name, col_type in zip(col_names, col_types, strict=True):
if col_type in numeric_types:
df[name] = pd.to_numeric(df[name], errors="coerce")
else:
df[name] = df[name].astype(object)
data[table] = df
conn.close()
return data
def _extract_from_csv_dir(self, csv_dir: Path) -> Dict[str, Any]:
"""Read all CSV files from *csv_dir* keyed by table name."""
import pandas as pd # lazy
csv_files = sorted(csv_dir.glob("*.csv"))
data: Dict[str, Any] = {}
for csv_file in csv_files:
table_name = csv_file.stem
df = pd.read_csv(
csv_file,
dtype=str,
keep_default_na=False,
na_values=[""],
)
data[table_name] = self._coerce_numeric_columns(df)
return self._order_data_by_schema(data)
def _order_data_by_schema(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Return *data* ordered using ORM metadata dependency order."""
ordered: Dict[str, Any] = {}
for table in Base.metadata.sorted_tables:
if table.name in data:
ordered[table.name] = data[table.name]
for name, frame in data.items():
if name not in ordered:
ordered[name] = frame
return ordered
def _finalize(
self,
data: Dict[str, Any],
*,
backend: str,
output_path: Optional[Path] = None,
) -> MigrationResult:
"""Create schema, load *data*, rename file, build the result.
When the engine points to a SQLite file (not ``:memory:``), the
file is moved to *output_path* if given, or otherwise renamed
to embed the release code and generation date into the
filename — e.g. ``dpm.db`` becomes ``dpm_4.2_20260512.db``.
The resulting path is returned in
:attr:`MigrationResult.database_path`. After the move the
original engine no longer points to a valid file — callers
that need to reuse the database must build a new engine from
the returned path.
"""
self._create_schema()
warnings = self._load_data(data)
table_details = {name: len(df) for name, df in data.items()}
total_rows = sum(table_details.values())
database_path = self._relocate_database(output_path)
return MigrationResult(
tables_migrated=len(table_details),
total_rows=total_rows,
table_details=table_details,
warnings=warnings,
backend_used=backend,
database_path=database_path,
)
# -------------------------------------------------------------- #
# Filename convention
# -------------------------------------------------------------- #
def _relocate_database(
self,
output_path: Optional[Path],
) -> Optional[Path]:
"""Move the SQLite file to its final location.
When *output_path* is given, the file is moved there verbatim;
otherwise the conventional ``<stem>_<release>_<YYYYMMDD>.db``
name is applied next to the original location. Returns
``None`` when the engine is not a SQLite file engine
(``:memory:``, PostgreSQL, etc.).
"""
current = self._sqlite_file_path()
if current is None:
return None
if output_path is not None:
new_path = Path(output_path)
else:
new_path = current.with_name(self._conventional_name(current))
if new_path == current:
return current
new_path.parent.mkdir(parents=True, exist_ok=True)
self._engine.dispose()
shutil.move(str(current), str(new_path))
return new_path
def _conventional_name(self, current: Path) -> str:
"""Build the ``<stem>_<release>_<YYYYMMDD><suffix>`` filename."""
tokens = [current.stem]
release_code = self._current_release_code()
if release_code:
tokens.append(release_code)
tokens.append(self._today_token())
return "_".join(tokens) + current.suffix
@staticmethod
def _today_token() -> str:
"""Return today's UTC date as ``YYYYMMDD``."""
return datetime.now(tz=timezone.utc).strftime("%Y%m%d")
def _sqlite_file_path(self) -> Optional[Path]:
"""Return the SQLite file path, or ``None`` if not applicable."""
url = self._engine.url
if url.get_backend_name() != "sqlite":
return None
database = url.database
if not database or database == ":memory:":
return None
path = Path(database)
if not path.is_file():
return None
return path
def _current_release_code(self) -> Optional[str]:
"""Return a filename-safe code for the current release."""
from sqlalchemy.orm import Session
from dpmcore.orm.release_sort_order import compute_sort_order
def _latest_code(rows: Any) -> Optional[str]:
# Latest first, including playground. release_id breaks ties.
ranked = sorted(
rows,
key=lambda r: (
compute_sort_order(r.date, r.type),
r.release_id,
),
reverse=True,
)
return ranked[0].code if ranked else None
with Session(self._engine) as session:
rows = session.execute(
select(
Release.code,
Release.date,
Release.type,
Release.release_id,
).where(Release.is_current == True) # noqa: E712
).all()
code = _latest_code(rows)
if code is None:
rows = session.execute(
select(
Release.code,
Release.date,
Release.type,
Release.release_id,
).where(Release.code.is_not(None))
).all()
code = _latest_code(rows)
if not code:
return None
return re.sub(r"[^A-Za-z0-9.+-]+", "-", code).strip("-") or None
@staticmethod
def _coerce_numeric_columns(df: Any) -> Any:
r"""Convert string columns to numeric where it is safe to do so.
``Code`` columns in the DPM dictionary store zero-padded
identifiers (``"0010"``, ``"010"``) that must not be coerced
to ints — doing so silently strips the padding and breaks
every downstream lookup.
A column is coerced only when:
- its name is not a known string-typed column
(``row``/``column``/``sheet``), and
- none of its non-null values start with a leading zero
followed by digits (``^0\d+``), and
- every non-null value can be parsed as numeric.
"""
import pandas as pd # lazy
string_columns = {"row", "column", "sheet"}
for column in df.columns:
if str(column).lower() in string_columns:
continue
non_null = df[column].dropna()
if non_null.empty:
continue
if non_null.astype(str).str.match(r"^0\d+").any():
continue
coerced = pd.to_numeric(non_null, errors="coerce")
if not coerced.isna().any():
df[column] = pd.to_numeric(df[column], errors="coerce")
return df
@staticmethod
def _coerce_temporal_columns_for_schema(df: Any, orm_table: Any) -> Any: # noqa: C901
"""Convert date/datetime columns to Python objects before loading."""
import pandas as pd # lazy
from sqlalchemy.sql.sqltypes import Date, DateTime
# fromisoformat() (Python 3.11+) covers all YYYY-MM-DD* variants;
# these fallbacks handle non-ISO formats from mdb-export and similar.
_DATE_FMTS = ("%d/%m/%Y", "%m/%d/%Y", "%m/%d/%y")
_DATETIME_FMTS = (
"%d/%m/%Y %H:%M:%S",
"%d/%m/%y %H:%M:%S",
"%m/%d/%Y %H:%M:%S",
"%m/%d/%y %H:%M:%S",
)
def is_missing(value: Any) -> bool:
try:
return (
value is None
or bool(pd.isna(value))
or (isinstance(value, str) and not value.strip())
)
except (TypeError, ValueError):
return False
def parse_date_value(value: Any) -> Optional[date]:
if is_missing(value):
return None
text = str(value).strip()
try:
return datetime.fromisoformat(text).date()
except ValueError:
pass
for fmt in _DATE_FMTS + _DATETIME_FMTS:
try:
return datetime.strptime(text, fmt).date() # noqa: DTZ007
except ValueError: # noqa: PERF203
continue
raise MigrationError(f"Unsupported date value {value!r}")
def parse_datetime_value(value: Any) -> Optional[datetime]:
if is_missing(value):
return None
text = str(value).strip()
try:
return datetime.fromisoformat(text)
except ValueError:
pass
for fmt in _DATETIME_FMTS:
try:
return datetime.strptime(text, fmt) # noqa: DTZ007
except ValueError: # noqa: PERF203
continue
for fmt in _DATE_FMTS:
try:
return datetime.combine(
datetime.strptime(text, fmt).date(), # noqa: DTZ007
datetime.min.time(),
)
except ValueError: # noqa: PERF203
continue
raise MigrationError(f"Unsupported datetime value {value!r}")
for column in orm_table.columns:
column_name = column.name
if column_name not in df.columns:
continue
if not isinstance(column.type, (Date, DateTime)):
continue
converted_values = []
bad_values = []
for raw_value in df[column_name].tolist():
try:
if isinstance(column.type, Date):
converted_values.append(parse_date_value(raw_value))
else:
converted_values.append(
parse_datetime_value(raw_value)
)
except MigrationError: # noqa: PERF203
bad_values.append(raw_value)
if bad_values:
unique_bad = list(
dict.fromkeys(
"<missing>" if is_missing(v) else repr(v)
for v in bad_values
)
)
raise MigrationError(
f"Table '{orm_table.name}', column '{column_name}' "
f"contains unsupported date values: {unique_bad[:5]}"
)
df[column_name] = pd.Series(converted_values, index=df.index)
return df
@staticmethod
def _coerce_boolean_columns_for_schema(df: Any, orm_table: Any) -> Any:
"""Convert Access-style boolean values to Python bools."""
from sqlalchemy.sql.sqltypes import Boolean
for column in orm_table.columns:
column_name = column.name
if column_name not in df.columns:
continue
if not isinstance(column.type, Boolean):
continue
try:
df[column_name] = df[column_name].map(
MigrationService._convert_bool_value
)
except MigrationError as exc:
raise MigrationError(
f"Table '{orm_table.name}', column '{column_name}' "
f"contains invalid boolean values: {exc}"
) from exc
return df
@staticmethod
def _convert_bool_value(value: Any) -> bool | None:
"""Coerce a single value to bool or None."""
import pandas as pd
if value is None or pd.isna(value):
return None
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
if value in {-1, 1}:
return True
if value == 0:
return False
return MigrationService._convert_text_bool(str(value).strip().lower())
@staticmethod
def _convert_text_bool(text: str) -> bool | None:
"""Coerce a normalised text token to bool or None."""
_null = {"", "nan", "none", "null", "<na>"}
_true = {"-1", "-1.0", "1", "1.0", "true", "yes", "y", "t"}
_false = {"0", "0.0", "false", "no", "n", "f"}
if text in _null:
return None
if text in _true:
return True
if text in _false:
return False
try:
numeric = float(text)
except ValueError as exc:
raise MigrationError(
f"Unsupported boolean value {text!r}"
) from exc
if numeric in {-1.0, 1.0}:
return True
if numeric == 0.0:
return False
raise MigrationError(f"Unsupported boolean value {text!r}")
def _prepare_bulk_load_constraints(self) -> None:
"""Prepare strict databases for bulk CSV loading."""
if self._schema is None:
return
dialect = self._engine.dialect.name
if dialect == "postgresql":
self._drop_postgresql_foreign_keys_for_bulk_load()
return
if dialect == "mssql":
self._disable_sqlserver_constraints_for_bulk_load()
return
def _drop_postgresql_foreign_keys_for_bulk_load(self) -> None:
"""Drop FK constraints in staging to allow bulk loading."""
query = text(
"""
SELECT table_name, constraint_name
FROM information_schema.table_constraints
WHERE table_schema = :schema
AND constraint_type = 'FOREIGN KEY'
"""
)
if self._schema is None:
return
preparer = self._engine.dialect.identifier_preparer
with self._engine.begin() as conn:
constraints = conn.execute(
query, {"schema": self._schema}
).fetchall()
for table_name, constraint_name in constraints:
qualified_table = (
f"{preparer.quote_schema(self._schema)}."
f"{preparer.quote(table_name)}"
)
conn.execute(
text(
f"ALTER TABLE {qualified_table} "
f"DROP CONSTRAINT {preparer.quote(constraint_name)}"
)
)
def _disable_sqlserver_constraints_for_bulk_load(self) -> None:
"""Disable FK/check constraints in staging to allow bulk loading."""
if self._schema is None:
return
inspector = inspect(self._engine)
table_names = inspector.get_table_names(schema=self._schema)
preparer = self._engine.dialect.identifier_preparer
with self._engine.begin() as conn:
for table_name in table_names:
qualified_table = (
f"{preparer.quote_schema(self._schema)}."
f"{preparer.quote(table_name)}"
)
conn.execute(
text(
f"ALTER TABLE {qualified_table} NOCHECK CONSTRAINT ALL"
)
)
# -------------------------------------------------------------- #
# Schema creation & data loading
# -------------------------------------------------------------- #
def _create_schema(self) -> None:
"""Drop and recreate all ORM tables for a clean migration."""
Base.metadata.drop_all(self._ddl_engine)
Base.metadata.create_all(self._ddl_engine)
self._prepare_bulk_load_constraints()
def _load_data(self, data: Dict[str, Any]) -> List[str]:
"""Write DataFrames into the database.
Uses ``if_exists="append"`` so that ORM-created column types
and constraints are preserved.
Returns:
A list of warning messages (e.g. tables that could not be
loaded).
"""
warnings: List[str] = []
identity_cols = (
self._mssql_identity_columns()
if self._engine.dialect.name == "mssql"
else {}
)
for table_name, df in data.items():
self._load_table(table_name, df, warnings, identity_cols)
self._resync_postgresql_sequences(data)
return warnings
def _resync_postgresql_sequences(self, data: Dict[str, Any]) -> None:
"""Resync SERIAL/IDENTITY sequences after loading explicit PKs.
``df.to_sql`` inserts every DataFrame column verbatim, including
single-column integer primary keys copied from the Access
source. Unlike SQL Server's ``IDENTITY_INSERT``, PostgreSQL
never advances a column's sequence when a value is supplied
explicitly, so a later ORM insert that relies on the sequence
default (e.g. a new ``OperationScope`` row) can collide with a
row that was just bulk-loaded. Resync each loaded table's
sequence to ``MAX(pk)`` once loading completes.
"""
if self._engine.dialect.name != "postgresql" or self._schema is None:
return
preparer = self._engine.dialect.identifier_preparer
with self._engine.begin() as conn:
for table_name in data:
orm_table = Base.metadata.tables.get(table_name)
if orm_table is None:
continue
pk_columns = list(orm_table.primary_key.columns)
if len(pk_columns) != 1:
continue
pk_column = pk_columns[0].name
quoted_pk_column = preparer.quote(pk_column)
qualified_table = (
f"{preparer.quote_schema(self._schema)}."
f"{preparer.quote(table_name)}"
)
sequence = conn.execute(
text("SELECT pg_get_serial_sequence(:table, :column)"),
{"table": qualified_table, "column": pk_column},
).scalar()
if sequence is None:
continue
max_id_sql = (
f"SELECT MAX({quoted_pk_column}) " # noqa: S608
f"FROM {qualified_table}"
)
max_id = conn.execute(text(max_id_sql)).scalar()
if max_id is None:
continue
conn.execute(
text("SELECT setval(:sequence, :max_id)"),
{"sequence": sequence, "max_id": max_id},
)
def _load_table(
self,
table_name: str,
df: Any,
warnings: List[str],
identity_columns: dict[str, str],
) -> None:
"""Load a single DataFrame into the database."""
# Filter DataFrame columns to only those present in the
# ORM schema so that unexpected Access columns produce a
# warning instead of a hard failure.
orm_table = Base.metadata.tables.get(table_name)
if orm_table is not None:
known_cols = {c.name for c in orm_table.columns}
extra_cols = set(df.columns) - known_cols
if extra_cols:
msg = (
f"Table '{table_name}': dropping unknown "
f"columns {sorted(extra_cols)}"
)
logger.info(msg)
warnings.append(msg)
df = df[[c for c in df.columns if c in known_cols]]
df = self._coerce_temporal_columns_for_schema(df, orm_table)
df = self._coerce_boolean_columns_for_schema(df, orm_table)
try:
identity_column = identity_columns.get(table_name)
if identity_column is not None and identity_column in df.columns:
preparer = self._engine.dialect.identifier_preparer
qualified_table = (
f"{preparer.quote_schema(self._schema)}."
f"{preparer.quote(table_name)}"
if self._schema is not None
else preparer.quote(table_name)
)
with self._engine.begin() as conn:
conn.execute(
text(f"SET IDENTITY_INSERT {qualified_table} ON")
)
try:
df.to_sql(
table_name,
conn,
schema=self._schema,
if_exists="append",
index=False,
chunksize=10_000,
)
finally:
conn.execute(
text(f"SET IDENTITY_INSERT {qualified_table} OFF")
)
else:
df.to_sql(
table_name,
self._engine,
schema=self._schema,
if_exists="append",
index=False,
chunksize=10_000,
)
except Exception as exc:
raise MigrationError(
f"Failed to load table '{table_name}': {exc}"
) from exc
def _mssql_identity_columns(self) -> dict[str, str]:
"""Return {table_name: identity_column} for tables in the schema."""
if self._schema is None:
return {}
with self._engine.connect() as conn:
rows = conn.execute(
text(
"""
SELECT t.name, c.name
FROM sys.columns c
JOIN sys.tables t ON c.object_id = t.object_id
JOIN sys.schemas s ON t.schema_id = s.schema_id
WHERE s.name = :schema
AND c.is_identity = 1
"""
),
{"schema": self._schema},
).fetchall()
return {row[0]: row[1] for row in rows}