Partial export

news/86.feature

@@ -0,0 +1 @@
+Export parts of the site specified on a file @gforcada @cihanandac

setup.py

@@ -85,6 +85,7 @@
         "z3c.autoinclude.plugin": ["target = plone"],
         "console_scripts": [
             "plone-exporter = plone.exportimport.cli:exporter_cli",
+            "plone-partial-exporter = plone.exportimport.cli:partial_exporter_cli",
             "plone-importer = plone.exportimport.cli:importer_cli",
         ],
     },

src/plone/exportimport/cli/__init__.py

@@ -1,3 +1,4 @@
+from pathlib import Path
 from plone import api
 from plone.exportimport import logger as package_logger
 from plone.exportimport.exporters import get_exporter
@@ -28,6 +29,16 @@
             "--quiet": "Do not report items being imported",
         },
     },
+    "partial-exporter": {
+        "description": "Export part of a Plone Site",
+        "options": {
+            "zopeconf": "Path to zope.conf",
+            "site": "Plone site ID or path to site to export the content from",
+            "objectpaths": "Path to the list of objects to export",
+            "path": "Path to export the content",
+            "--include-revisions": "Include revision history",
+        },
+    },
 }
 
 
@@ -61,6 +72,46 @@ def exporter_cli(args=sys.argv):
         logger.info(f" Wrote {item}")
 
 
+def partial_exporter_cli(args=sys.argv):
+    """Export part of a Plone site."""
+    logger = cli_helpers.get_logger("Partial exporter")
+    exporter_cli = CLI_SPEC["partial-exporter"]
+    # We get an argparse.Namespace instance.
+    namespace = _parse_args(exporter_cli["description"], exporter_cli["options"], args)
+    app = cli_helpers.get_app(namespace.zopeconf)
+    path = cli_helpers._process_path(namespace.path)
+    if not path:
+        logger.error(f"{namespace.path} does not exist, please create it first.")
+        sys.exit(1)
+    object_paths = cli_helpers._process_path(namespace.objectpaths)
+    if not object_paths:
+        logger.error(f"{namespace.objectpaths} file does not exist, create it.")
+        sys.exit(1)
+    raw_paths = Path(object_paths).read_text().splitlines()
+    site = cli_helpers.get_site(app, namespace.site, logger)
+
+    # Add site id to the beginning of the paths if it is not already present
+    paths_list = []
+    for raw_path in raw_paths:
+        raw_path = raw_path.strip()
+        if not raw_path:
+            continue
+        raw_path = raw_path.lstrip("/")
+        if not (raw_path == site.id or raw_path.startswith(f"{site.id}/")):
+            raw_path = f"{site.id}/{raw_path}"
+        paths_list.append(f"/{raw_path}")
+
+    with api.env.adopt_roles(["Manager"]):
+        results = get_exporter(site).partial_export_site(
+            path, paths_list=paths_list, options=namespace
+        )
+    logger.info(
+        f" Using path {path} to export partial content from Plone site at /{site.id}"
+    )
+    for item in results[1:]:
+        logger.info(f" Wrote {item}")
+
+
 def importer_cli(args=sys.argv):
     """Import content to a Plone site."""
     logger = cli_helpers.get_logger("Importer")

src/plone/exportimport/exporters/__init__.py

@@ -73,6 +73,22 @@ def export_site(
                 paths.extend(new_paths)
         return paths
 
+    def partial_export_site(
+        self,
+        path: Optional[Path] = None,
+        paths_list: Optional[List[str]] = [],
+        options: Optional[argparse.Namespace] = None,
+    ) -> List[Path]:
+        """Export the given site to the filesystem."""
+        path = self._prepare_path(path)
+        paths: List[Path] = [path]
+        with hooks.site(self.site):
+            for exporter_name, exporter in self.exporters.items():
+                logger.debug(f"Exporting {self.site} with {exporter_name} to {path}")
+                new_paths = exporter.export_data(path, paths_list, options=options)
+                paths.extend(new_paths)
+        return paths
+
 
 def get_exporter(site: PloneSite = None) -> Exporter:
     """Get the exporter."""

src/plone/exportimport/exporters/content.py

@@ -21,6 +21,7 @@
 class ContentExporter(BaseExporter):
     name: str = "content"
     query: dict = None
+    paths_list: list[str] = []
     filename_fmt: str = settings.EXPORT_CONTENT_FILEPATH
     metadata: types.ExportImportMetadata = None
     default_site_language: str = "en"
@@ -132,19 +133,51 @@ def dump_metadata(self) -> Path:
 
     def dump(self) -> list[Path]:
         """Serialize contents and dump them to disk."""
-        paths = []
         with request_provides(self.request, IExportImportRequestMarker):
-            for obj in self.all_objects():
-                path = self.dump_one(obj)
-                if path:
-                    paths.append(path)
-            # Add list of blobs to serialization
-            paths.insert(0, self.dump_metadata())
+            if not self.paths_list:
+                return self._dump_all()
+            return self._dump_selected()
+
+    def get_selected_objects(self) -> Generator:
+        """Return all selected objects to be serialized."""
+        logger.info(f"Exporting {len(self.paths_list)}")
+        for index, path in enumerate(self.paths_list, start=1):
+            try:
+                obj = content_utils.object_from_path(path)
+            except Exception:
+                msg = f"Error getting object at path {path}"
+                self.errors.append({"path": path, "message": msg})
+                logger.exception(msg, exc_info=True)
+            else:
+                yield obj
+
+            if not index % 100:
+                logger.info(f"Content Exporter: Handled {index} items...")
+
+    def _dump_all(self) -> List[Path]:
+        paths = []
+        for obj in self.all_objects():
+            path = self.dump_one(obj)
+            if path:
+                paths.append(path)
+        # Add list of blobs to serialization
+        paths.insert(0, self.dump_metadata())
+        return paths
+
+    def _dump_selected(self) -> List[Path]:
+        paths = []
+        for obj in self.get_selected_objects():
+            path = self.dump_one(obj)
+            if path and path not in paths:
+                paths.append(path)
+        # Add list of blobs to serialization
+        paths.insert(0, self.dump_metadata())
         return paths
 
     def export_data(
         self,
         base_path: Path,
+        paths_list: list[str] | None = [],
         data_hooks: list[Callable] = None,
         obj_hooks: list[Callable] = None,
         query: dict | None = None,
@@ -154,6 +187,7 @@ def export_data(
         base_path = base_path / self.name
         query = query if query else {}
         site = self.site
+        self.paths_list = paths_list
         self.query = query if query else {"path": content_utils.get_obj_path(site)}
         metadata = types.ExportImportMetadata()
         self.metadata = metadata

src/plone/exportimport/utils/content/__init__.py

@@ -4,6 +4,7 @@
 from .core import get_uid  # noQA
 from .core import is_folderish  # noQA
 from .core import is_site_root  # noQA
+from .core import object_from_path  # noQA
 from .core import object_from_uid  # noQA
 from .export_helpers import cleaners  # noQA
 from .export_helpers import enrichers  # noQA

src/plone/exportimport/utils/content/core.py

@@ -65,6 +65,17 @@ def object_from_uid_or_path(uid: str, path: str = "") -> DexterityContent | None
     return obj
 
 
+def object_from_path(path: str) -> Optional[DexterityContent]:
+    """Return an object for a given path."""
+    obj = None
+    # Try first to get the object by its path
+    portal: PloneSite = api.portal.get()
+    obj = portal.unrestrictedTraverse(path, default=None)
+    if not obj:
+        raise ValueError(f"Object not found at path: {path}")
+    return obj
+
+
 def get_portal_languages() -> types.PortalLanguages:
     """Return configured languages in a Plone Site."""
     default = api.portal.get_registry_record("plone.default_language", default="en")

tests/exporters/test_exporters_partial.py

@@ -0,0 +1,88 @@
+from pathlib import Path
+from plone.exportimport.exporters import content
+
+import pytest
+
+# objects in `base_import`
+FOO = "e7359727ace64e609b79c4091c38822a"  # /plone/foo
+FOO_BAR = "35661c9bb5be42c68f665aa1ed291418"  # /plone/foo/bar
+FOO_ANOTHER = "45b0b46f17104a7b8fa7bb94d3dd5bd9"  # /plone/foo/another-page
+BAR = "70844f7bec1843b8ab2796c972c9ebfe"  # /plone/bar
+BAR_NEWS = "7c1393f615c4447c80db0d784390c5b7"  # /plone/bar/an-important-news
+SITE_ROOT = "plone_site_root"  # /plone
+
+FOO_PATH = "/plone/foo"
+FOO_BAR_PATH = "/plone/foo/bar"
+BAR_NEWS_PATH = "/plone/bar/an-important-news"
+
+
+def exported_uids(export_path: Path) -> set:
+    """Return the set of UID directories that hold a serialized object."""
+    content_dir = Path(export_path) / "content"
+    if not content_dir.is_dir():
+        return set()
+    return {path.parent.name for path in content_dir.glob("*/data.json")}
+
+
+class TestPartialContentExporter:
+    """The content exporter only serializes the explicitly selected objects."""
+
+    @pytest.fixture(autouse=True)
+    def _init(self, portal):
+        self.portal = portal
+        self.exporter = content.ContentExporter(portal)
+
+    def test_only_selected_object_is_exported(self, export_path):
+        # Only includes the objects specified in the paths list.
+        self.exporter.export_data(
+            base_path=export_path, paths_list=[FOO_PATH, BAR_NEWS_PATH]
+        )
+        assert exported_uids(export_path) == {FOO, BAR_NEWS}
+
+    def test_children_objects_are_not_exported(self, export_path):
+        # Unless the export paths include the child paths, they are not exported.
+        self.exporter.export_data(base_path=export_path, paths_list=[FOO_PATH])
+        uids = exported_uids(export_path)
+        assert FOO_BAR not in uids
+        assert FOO_ANOTHER not in uids
+
+    def test_sibling_objects_are_not_exported(self, export_path):
+        # Unless the export paths include the sibling object paths, they are not exported.
+        self.exporter.export_data(base_path=export_path, paths_list=[FOO_PATH])
+        assert BAR not in exported_uids(export_path)
+
+    def test_parent_and_root_are_not_exported(self, export_path):
+        # Selecting a child must not drag in its parent or the site root.
+        self.exporter.export_data(base_path=export_path, paths_list=[FOO_BAR_PATH])
+        uids = exported_uids(export_path)
+        assert uids == {FOO_BAR}
+        assert FOO not in uids
+        assert SITE_ROOT not in uids
+
+    def test_metadata_lists_only_selected(self, export_path, load_json):
+        # The metadata should only list the selected objects.
+        self.exporter.export_data(base_path=export_path, paths_list=[FOO_PATH])
+        metadata = load_json(export_path, "content/__metadata__.json")
+        assert metadata["_data_files_"] == [f"{FOO}/data.json"]
+
+    def test_duplicate_paths_are_written_once(self, export_path, load_json):
+        # Even if a path is specified multiple times, it should only be written once.
+        self.exporter.export_data(
+            base_path=export_path, paths_list=[FOO_PATH, FOO_PATH]
+        )
+        assert exported_uids(export_path) == {FOO}
+        metadata = load_json(export_path, "content/__metadata__.json")
+        assert metadata["_data_files_"] == [f"{FOO}/data.json"]
+
+    def test_unknown_path_is_skipped(self, export_path):
+        # An unresolvable paths are skipped; valid paths are still exported.
+        self.exporter.export_data(
+            base_path=export_path, paths_list=["/plone/does-not-exist", FOO_PATH]
+        )
+        assert exported_uids(export_path) == {FOO}
+
+    def test_empty_paths_list_exports_everything(self, export_path):
+        # An empty list behaves like a full export.
+        self.exporter.export_data(base_path=export_path, paths_list=[])
+        uids = exported_uids(export_path)
+        assert {FOO, BAR, FOO_BAR, BAR_NEWS, SITE_ROOT} <= uids