dpmcore.services.database_update

Safely replace an existing DPM database with fresh data. Data is loaded into a temporary staging area and validated before the active database is touched; if anything fails the original is left intact.

Note

Requires the migration extra: pip install dpmcore[migration].

DatabaseUpdateService

class dpmcore.services.database_update.DatabaseUpdateService[source]

Bases: object

Safely update DPM databases.

update(*, target, access_file=None, ecb_validations_file=None, source_dir='data/DPM', dry_run=False, keep_staging=False)[source]

Update a target database from CSVs or an Access file.

Return type:

DatabaseUpdateResult

Parameters:

  • target (str)

  • access_file (str | None)

  • ecb_validations_file (str | None)

  • source_dir (str)

  • dry_run (bool)

  • keep_staging (bool)

static detect_target_type(target)[source]

Detect target type from URL or file path.

Return type:

str

Parameters:

target (str)

DatabaseUpdateResult

class dpmcore.services.database_update.DatabaseUpdateResult(target_type, target, source, used_access_file, migration_result, ecb_validations_imported, dry_run=False, staging_location=None)[source]

Result of a safe database update.

Parameters:
target_type: str
target: str
source: str
used_access_file: bool
migration_result: MigrationResult
ecb_validations_imported: bool
dry_run: bool = False
staging_location: str | None = None

DatabaseUpdateError

exception dpmcore.services.database_update.DatabaseUpdateError[source]

Bases: Exception

Raised when a safe database update cannot be completed.