feat: add custom metadata support (#663)

Added support for custom metadata writing and validation during operations via `DBBACKUP_BACKUP_METADATA_SETTER` and `DBBACKUP_RESTORE_METADATA_VALIDATOR` settings.

Author: matmair

CHANGELOG.md

@@ -14,7 +14,8 @@ Don't forget to remove deprecated code on each major release!
 
 ## [Unreleased]
 
-- Nothing (yet)!
+### Added
+- Added support for custom metadata writing and validation during operations via `DBBACKUP_BACKUP_METADATA_SETTER` and `DBBACKUP_RESTORE_METADATA_VALIDATOR` settings.
 
 ## [5.1.2] - 2026-01-14
 

dbbackup/management/commands/dbbackup.py

@@ -101,6 +101,11 @@ def _save_metadata(self, filename, local=False):
             "connector": f"{self.connector.__module__}.{self.connector.__class__.__name__}",
         }
         metadata_filename = f"{filename}.metadata"
+
+        # Load custom metadata if configured
+        user_metadata = utils.get_user_metadata(metadata)
+        metadata.update(user_metadata)
+
         metadata_content = json.dumps(metadata)
 
         if local:

dbbackup/management/commands/dbrestore.py

@@ -153,6 +153,12 @@ def _check_metadata(self, filename):
             )
             raise CommandError(msg)
 
+        # Check if we custom metadata validation is configured
+        user_metadata_valid = utils.validate_user_metadata(metadata)
+        if not user_metadata_valid:
+            msg = f"Custom metadata validation failed for backup file '{filename}'."
+            raise CommandError(msg)
+
         return metadata
 
     def _restore_backup(self):

dbbackup/settings.py

@@ -42,3 +42,5 @@
 SERVER_EMAIL = getattr(settings, "DBBACKUP_SERVER_EMAIL", settings.SERVER_EMAIL)
 ADMINS = getattr(settings, "DBBACKUP_ADMIN", settings.ADMINS)
 EMAIL_SUBJECT_PREFIX = getattr(settings, "DBBACKUP_EMAIL_SUBJECT_PREFIX", "[dbbackup] ")
+BACKUP_METADATA_SETTER = getattr(settings, "DBBACKUP_BACKUP_METADATA_SETTER", None)
+RESTORE_METADATA_VALIDATOR = getattr(settings, "DBBACKUP_RESTORE_METADATA_VALIDATOR", None)

dbbackup/utils.py

@@ -2,7 +2,11 @@
 Utility functions for dbbackup.
 """
 
+from __future__ import annotations
+
+import copy
 import gzip
+import json
 import logging
 import os
 import re
@@ -12,6 +16,7 @@
 from datetime import datetime
 from functools import wraps
 from getpass import getpass
+from importlib import import_module
 from shutil import copyfileobj
 
 from django.core.mail import EmailMultiAlternatives
@@ -431,3 +436,93 @@ def filename_generate(extension, database_name="", servername=None, content_type
         filename = REG_FILENAME_CLEAN.sub("-", filename)
         filename = filename.removeprefix("-")
     return filename
+
+
+def _get_function_from_path(func_or_path):
+    """
+    Load a callable from a dotted path or return the callable itself.
+
+    :param func_or_path: Dotted path to callable or callable itself
+    :type func_or_path: str | callable
+
+    :returns: Callable object
+    :rtype: callable
+    """
+    if callable(func_or_path):
+        return func_or_path
+
+    path = func_or_path
+    module_path, func_name = path.rsplit(".", 1)
+    try:
+        module = import_module(module_path)
+    except ImportError as e:
+        msg = f"Could not import module '{module_path}': {e}"
+        raise ImportError(msg) from e
+    func = getattr(module, func_name)
+    if not callable(func):
+        msg = f"The object at '{path}' is not callable."
+        raise TypeError(msg)
+    return func
+
+
+def get_user_metadata(metadata: dict | None = None) -> dict:
+    """
+    Get user generated metadata from the user's custom metadata setter.
+
+    :returns: Custom metadata dictionary
+    :rtype: dict
+    """
+    user_metadata = {}
+    setter_setting = settings.BACKUP_METADATA_SETTER
+    if setter_setting:
+        setter_function = _get_function_from_path(setter_setting)
+        try:
+            # We pass a copy to avoid side effects
+            user_metadata = setter_function(copy.deepcopy(metadata))
+        except Exception:
+            logger = logging.getLogger("dbbackup")
+            logger.exception("Error loading custom metadata: %s")
+
+        if user_metadata is None:
+            user_metadata = {}
+
+        if not isinstance(user_metadata, dict):
+            msg = "DBBACKUP_BACKUP_METADATA_SETTER must return a dictionary."
+            raise ValueError(msg)
+
+        # Validate that we can serialize the provided data
+        try:
+            json.dumps(user_metadata)
+        except Exception as e:
+            msg = f"Custom metadata is not JSON serializable: {e}"
+            raise ValueError(msg) from e
+    return user_metadata
+
+
+def validate_user_metadata(metadata) -> bool | None:
+    """
+    Validate custom metadata using a callable defined in `settings.py`.
+    Raise a CommandError to provide custom feedback if validation fails.
+
+    :param metadata: Metadata dictionary to validate
+    :type metadata: dict
+
+    :returns: True if validation passes (or None), False otherwise
+    :rtype: Optional[bool]
+    """
+    validator_setting = settings.RESTORE_METADATA_VALIDATOR
+    if validator_setting:
+        validator_function = _get_function_from_path(validator_setting)
+        try:
+            # We pass a copy to avoid side effects
+            user_metadata = validator_function(copy.deepcopy(metadata))
+        except Exception as e:
+            msg = f"Error during custom metadata validation: {e}"
+            raise ValueError(msg) from e
+        if user_metadata is None:
+            return None
+        if not isinstance(user_metadata, bool):
+            msg = "DBBACKUP_RESTORE_METADATA_VALIDATOR must return a boolean or None."
+            raise TypeError(msg)
+        return user_metadata
+    return True

docs/src/configuration.md

@@ -123,9 +123,9 @@ python manage.py dbrestore --decrypt
 
 Requirements:
 
--   Install the python package python-gnupg: `pip install python-gnupg>=0.5.0`.
--   You need a GPG key. ([GPG manual](https://www.gnupg.org/gph/en/manual/c14.html))
--   Set the setting `DBBACKUP_GPG_RECIPIENT` to the name of the GPG key.
+- Install the python package python-gnupg: `pip install python-gnupg>=0.5.0`.
+- You need a GPG key. ([GPG manual](https://www.gnupg.org/gph/en/manual/c14.html))
+- Set the setting `DBBACKUP_GPG_RECIPIENT` to the name of the GPG key.
 
 Note (Windows): The `gpg` executable must be installed and on your PATH for encryption/decryption. If it is absent, django-dbbackup still works; only encryption-related features are unavailable. The test suite will automatically skip encryption tests when `gpg` is not found.
 
@@ -204,3 +204,53 @@ By default DBBackup uses values from `settings.DATABASES`. Use
 
 You must configure a storage backend (`STORAGES['dbbackup']`) to persist
 backups. See [Storage settings](storage.md) for supported options.
+
+## Custom metadata
+
+### DBBACKUP_BACKUP_METADATA_SETTER
+
+Function or dotted path (string) to a callable that returns a dictionary of extra metadata.
+
+The function must accept a single argument: a metadata dictionary containing information about the current backup operation. If a dictionary is returned, it will be merged into the current backup's metadata file. If `None` is returned, it is treated as an empty dictionary.
+
+Default: `None`
+
+```python
+def metadata_set(metadata):
+    last_backup_time = datetime.now().isoformat()
+    if metadata and metadata.get('engine') == 'my.custom.Engine':
+        region = os.environ.get('AWS_REGION')
+        return {'environment': 'AWS', 'region': region, 'backup_time': last_backup_time}
+    else:
+        return {'backup_time': last_backup_time}
+
+DBBACKUP_BACKUP_METADATA_SETTER = metadata_set
+```
+
+### DBBACKUP_RESTORE_METADATA_VALIDATOR
+
+Function or dotted path (string) to a callable that performs additional validation on the backup's metadata during restore operations. This validator runs **after** the built-in validation (e.g. database engine checks).
+
+The callable should accept a single argument: the metadata dictionary loaded from the current restore operation's metadata file. If this function returns `True`, the metadata is valid, `False` means the restore operation will be aborted. `None` can be returned to do nothing and allow `dbbackup` to decide. It may raise a `CommandError` with a descriptive message if validation fails.
+
+Default: `None`
+
+```python
+import os
+
+def validate_restore(metadata):
+    region = os.environ.get('AWS_REGION')
+    if not metadata.get('region') == region:
+        raise CommandError(f"Backup region does not match current region {region}; cross region restores are not allowed due to SMP-0123.")
+
+    last_backup_time = metadata.get('backup_time')
+    if datetime.now() - datetime.fromisoformat(last_backup_time) > timedelta(days=120):
+        return False # Skip restore since it is stale (>120 days)
+
+    if os.environ.get('AWS_REGION') == None:
+        return None # Do nothing if performing a backup outside AWS_REGION
+
+    return True
+
+DBBACKUP_RESTORE_METADATA_VALIDATOR = validate_restore
+```

docs/src/dictionary.txt

@@ -48,3 +48,5 @@ postgis
 postgresql
 gzip
 tarfile
+validator
+no-op

pyproject.toml

@@ -119,6 +119,8 @@ matrix.django.dependencies = [
 # >>> Documentation Scripts <<<
 
 [tool.hatch.envs.docs]
+installer = "uv"
+python = "3.11" # `editdistpy` (v0.1.6) is a dependency of `mkdocs-spellcheck[all]`, and is incompatible with Python 3.12+.
 template = "docs"
 detached = true
 dependencies = [

tests/commands/test_dbrestore_metadata.py

@@ -1,11 +1,14 @@
+import importlib
 import json
 from unittest.mock import Mock, patch
 
 import pytest
 from django.conf import settings
 from django.core.management.base import CommandError
 from django.test import TestCase
+from django.test.utils import override_settings
 
+from dbbackup.management.commands import dbbackup
 from dbbackup.management.commands.dbrestore import Command as DbrestoreCommand
 
 
@@ -83,6 +86,55 @@ def test_django_connector_mismatch_allowed(self):
         # Should not raise
         self.command._check_metadata("backup.dump")
 
+    @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator")
+    def test_metadata_match_custom(self):
+        importlib.reload(dbbackup.settings)
+        # Setup metadata
+        metadata = {"engine": settings.DATABASES["default"]["ENGINE"], "CSMT_VAL": "aabbcc-1122-3344_eu-west"}
+        self.command.storage.read_file.return_value = Mock(read=lambda: json.dumps(metadata))
+
+        # Should not raise
+        self.command._check_metadata("backup.dump")
+
+    @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator")
+    def test_metadata_match_custom_fail(self):
+        importlib.reload(dbbackup.settings)
+        # Setup metadata
+        metadata = {"engine": settings.DATABASES["default"]["ENGINE"], "CSMT_VAL": "aabbcc-1122-3344_eu-ttt"}
+        self.command.storage.read_file.return_value = Mock(read=lambda: json.dumps(metadata))
+
+        # Should raise CommandError due to validation failure
+        with pytest.raises(CommandError):
+            self.command._check_metadata("backup.dump")
+
+    @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator")
+    def test_metadata_match_custom_failwithcustomerror(self):
+        importlib.reload(dbbackup.settings)
+        # Setup metadata
+        metadata = {"engine": settings.DATABASES["default"]["ENGINE"], "CSMT_VAL": "xx-1122-3344_eu-west"}
+        self.command.storage.read_file.return_value = Mock(read=lambda: json.dumps(metadata))
+
+        # Should raise CommandError due to validation failure
+        with pytest.raises(ValueError, match="CSMT_VAL must start with 'aabbcc'"):
+            self.command._check_metadata("backup.dump")
+
+    @override_settings(DBBACKUP_RESTORE_METADATA_VALIDATOR="tests.test_user_metadata.dummy_validator")
+    def test_metadata_mismatch_custom_valid(self):
+        """Test that built-in validation runs before custom validation."""
+        importlib.reload(dbbackup.settings)
+        # Setup metadata with different engine (primary fail) but valid custom data
+        metadata = {
+            "engine": "django.db.backends.postgresql",
+            "CSMT_VAL": "aabbcc-1122-3344_eu-west"
+        }
+        self.command.storage.read_file.return_value = Mock(read=lambda: json.dumps(metadata))
+
+        # Should raise CommandError due to engine mismatch (primary), not custom validation
+        with pytest.raises(CommandError) as cm:
+            self.command._check_metadata("backup.dump")
+
+        assert "Restoring to a different database engine is not supported" in str(cm.value)
+
 
 class DbrestoreConnectorOverrideTest(TestCase):
     def setUp(self):