GovOPlaN/govoplan-files
1
0
Fork 0
Code Issues 13 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(files): decompose campaign snapshot preparation

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 12:57:40 +02:00
parent 92950af6f4
commit 3449cbc8a5
1 changed files with 196 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

294
src/govoplan_files/backend/storage/campaign_attachments.py
View File

@@ -12,8 +12,8 @@ from typing import Any, Iterator
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from govoplan_files.backend.db.models import FileAsset, FileShare from govoplan_files.backend.db.models import FileAsset, FileBlob, FileShare, FileVersion
from govoplan_files.backend.storage.backends import StorageBackendError, get_storage_backend from govoplan_files.backend.storage.backends import StorageBackend, StorageBackendError, get_storage_backend
from govoplan_files.backend.storage.common import FileStorageError from govoplan_files.backend.storage.common import FileStorageError
from govoplan_files.backend.storage.files import current_versions_and_blobs, get_asset_for_user, list_assets_for_user, share_files from govoplan_files.backend.storage.files import current_versions_and_blobs, get_asset_for_user, list_assets_for_user, share_files
from govoplan_files.backend.storage.access import ensure_owner_access from govoplan_files.backend.storage.access import ensure_owner_access
@@ -232,6 +232,177 @@ def _selected_base_path(
return None return None
def _prepared_attachment_config(raw_json: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any], list[Any]]:
prepared_json = copy.deepcopy(raw_json if isinstance(raw_json, dict) else {})
attachments = prepared_json.get("attachments")
if not isinstance(attachments, dict):
attachments = {}
prepared_json["attachments"] = attachments
base_paths = attachments.get("base_paths")
if not isinstance(base_paths, list):
base_paths = []
return prepared_json, attachments, base_paths
def _campaign_snapshot_assets(
session: Session,
*,
tenant_id: str,
campaign_id: str,
prepared_json: dict[str, Any],
include_unlinked_candidates: bool,
user_id: str,
is_admin: bool,
) -> tuple[list[FileAsset], list[FileAsset], set[str]]:
shared_assets = list_assets_for_user(
session,
tenant_id=tenant_id,
user_id="",
campaign_id=campaign_id,
is_admin=True,
)
if not include_unlinked_candidates:
return shared_assets, shared_assets, {asset.id for asset in shared_assets}
candidate_assets, linked_asset_ids = _candidate_assets_for_managed_sources(
session,
tenant_id=tenant_id,
campaign_id=campaign_id,
raw_json=prepared_json,
user_id=user_id,
is_admin=is_admin,
shared_assets=shared_assets,
)
return shared_assets, candidate_assets, linked_asset_ids
def _assets_grouped_by_owner(assets: list[FileAsset]) -> dict[tuple[str, str], list[FileAsset]]:
assets_by_owner: dict[tuple[str, str], list[FileAsset]] = defaultdict(list)
for asset in assets:
owner_id = _asset_owner_id(asset)
if owner_id:
assets_by_owner[(asset.owner_type, owner_id)].append(asset)
return assets_by_owner
def _materialize_managed_asset(
asset: FileAsset,
*,
owner_id: str,
logical_root: str,
local_root: Path,
version_blobs: dict[str, tuple[FileVersion, FileBlob]],
backend: StorageBackend | None,
include_bytes: bool,
linked_asset_ids: set[str],
) -> tuple[str, ManagedAttachmentFile] | None:
relative_path = _relative_asset_path(asset, logical_root)
if not relative_path:
return None
target = _safe_local_target(local_root, relative_path)
target.parent.mkdir(parents=True, exist_ok=True)
version, blob = version_blobs[asset.id]
if include_bytes:
try:
data = backend.get_bytes(blob.storage_key) if backend else b""
except StorageBackendError as exc:
raise FileStorageError(str(exc)) from exc
target.write_bytes(data)
else:
target.touch()
local_key = str(target.resolve())
return local_key, ManagedAttachmentFile(
local_path=local_key,
asset_id=asset.id,
version_id=version.id,
blob_id=blob.id,
display_path=asset.display_path,
relative_path=normalize_logical_path(relative_path),
filename=asset.filename,
owner_type=asset.owner_type,
owner_id=owner_id,
checksum_sha256=blob.checksum_sha256,
size_bytes=blob.size_bytes,
content_type=blob.content_type,
linked_to_campaign=asset.id in linked_asset_ids,
source_provenance=source_provenance_from_metadata(asset.metadata_),
source_revision=source_revision_from_metadata(asset.metadata_),
)
def _prepare_managed_base_paths(
base_paths: list[Any],
*,
materialized_root: Path,
assets_by_owner: dict[tuple[str, str], list[FileAsset]],
version_blobs: dict[str, tuple[FileVersion, FileBlob]],
backend: StorageBackend | None,
include_bytes: bool,
linked_asset_ids: set[str],
) -> tuple[
dict[str, ManagedAttachmentFile],
dict[str, tuple[str, str]],
dict[str, list[tuple[str, str]]],
tuple[str, str] | None,
]:
manifest: dict[str, ManagedAttachmentFile] = {}
prepared_by_id: dict[str, tuple[str, str]] = {}
prepared_by_old_path: dict[str, list[tuple[str, str]]] = {}
first_prepared: tuple[str, str] | None = None
for index, item in enumerate(base_paths):
if not isinstance(item, dict):
continue
parsed_source = parse_managed_source(item.get("source"))
if parsed_source is None:
continue
owner_type, owner_id = parsed_source
old_path = str(item.get("path") or ".").strip() or "."
logical_root = "" if old_path in {"", ".", "/"} else normalize_folder(old_path)
base_path_id = str(item.get("id") or f"base-path-{index + 1}")
local_root = materialized_root / f"{index + 1:03d}-{safe_storage_component(base_path_id)}"
local_root.mkdir(parents=True, exist_ok=True)
local_root_string = str(local_root.resolve())
prepared = (base_path_id, local_root_string)
prepared_by_id[base_path_id] = prepared
prepared_by_old_path.setdefault(old_path, []).append(prepared)
if first_prepared is None:
first_prepared = prepared
item["path"] = local_root_string
for asset in assets_by_owner.get((owner_type, owner_id), []):
materialized = _materialize_managed_asset(
asset,
owner_id=owner_id,
logical_root=logical_root,
local_root=local_root,
version_blobs=version_blobs,
backend=backend,
include_bytes=include_bytes,
linked_asset_ids=linked_asset_ids,
)
if materialized is not None:
local_key, managed_file = materialized
manifest[local_key] = managed_file
return manifest, prepared_by_id, prepared_by_old_path, first_prepared
def _rewrite_managed_attachment_rules(
attachments: dict[str, Any],
prepared_json: dict[str, Any],
*,
prepared_by_id: dict[str, tuple[str, str]],
prepared_by_old_path: dict[str, list[tuple[str, str]]],
first_prepared: tuple[str, str] | None,
) -> None:
for rule in _iter_rule_dicts(attachments, prepared_json):
selected = _selected_base_path(rule, prepared_by_id, prepared_by_old_path, first_prepared)
if selected is None:
continue
base_path_id, local_root_string = selected
rule["base_path_id"] = base_path_id
rule["base_dir"] = local_root_string
if first_prepared is not None:
attachments["base_path"] = first_prepared[1]
def prepare_campaign_snapshot( def prepare_campaign_snapshot(
session: Session, session: Session,
*, *,
@@ -258,114 +429,35 @@ def prepare_campaign_snapshot(
materialized_root = destination / "managed-attachments" materialized_root = destination / "managed-attachments"
materialized_root.mkdir(parents=True, exist_ok=True) materialized_root.mkdir(parents=True, exist_ok=True)
prepared_json = copy.deepcopy(raw_json if isinstance(raw_json, dict) else {}) prepared_json, attachments, base_paths = _prepared_attachment_config(raw_json)
attachments = prepared_json.get("attachments") shared_assets, candidate_assets, linked_asset_ids = _campaign_snapshot_assets(
if not isinstance(attachments, dict):
attachments = {}
prepared_json["attachments"] = attachments
base_paths = attachments.get("base_paths")
if not isinstance(base_paths, list):
base_paths = []
shared_assets = list_assets_for_user(
session,
tenant_id=tenant_id,
user_id="",
campaign_id=campaign_id,
is_admin=True,
)
candidate_assets = shared_assets
linked_asset_ids = {asset.id for asset in shared_assets}
if include_unlinked_candidates:
candidate_assets, linked_asset_ids = _candidate_assets_for_managed_sources(
session, session,
tenant_id=tenant_id, tenant_id=tenant_id,
campaign_id=campaign_id, campaign_id=campaign_id,
raw_json=prepared_json, prepared_json=prepared_json,
include_unlinked_candidates=include_unlinked_candidates,
user_id=user_id, user_id=user_id,
is_admin=is_admin, is_admin=is_admin,
shared_assets=shared_assets,
) )
assets_by_owner = _assets_grouped_by_owner(candidate_assets)
assets_by_owner: dict[tuple[str, str], list[FileAsset]] = defaultdict(list)
for asset in candidate_assets:
owner_id = _asset_owner_id(asset)
if owner_id:
assets_by_owner[(asset.owner_type, owner_id)].append(asset)
version_blobs = current_versions_and_blobs(session, candidate_assets) version_blobs = current_versions_and_blobs(session, candidate_assets)
backend = get_storage_backend() if include_bytes else None backend = get_storage_backend() if include_bytes else None
manifest, prepared_by_id, prepared_by_old_path, first_prepared = _prepare_managed_base_paths(
manifest: dict[str, ManagedAttachmentFile] = {} base_paths,
prepared_by_id: dict[str, tuple[str, str]] = {} materialized_root=materialized_root,
prepared_by_old_path: dict[str, list[tuple[str, str]]] = {} assets_by_owner=assets_by_owner,
first_prepared: tuple[str, str] | None = None version_blobs=version_blobs,
backend=backend,
for index, item in enumerate(base_paths): include_bytes=include_bytes,
if not isinstance(item, dict): linked_asset_ids=linked_asset_ids,
continue )
parsed_source = parse_managed_source(item.get("source")) _rewrite_managed_attachment_rules(
if parsed_source is None: attachments,
continue prepared_json,
owner_type, owner_id = parsed_source prepared_by_id=prepared_by_id,
old_path = str(item.get("path") or ".").strip() or "." prepared_by_old_path=prepared_by_old_path,
logical_root = "" if old_path in {"", ".", "/"} else normalize_folder(old_path) first_prepared=first_prepared,
base_path_id = str(item.get("id") or f"base-path-{index + 1}")
local_root = materialized_root / f"{index + 1:03d}-{safe_storage_component(base_path_id)}"
local_root.mkdir(parents=True, exist_ok=True)
local_root_string = str(local_root.resolve())
prepared = (base_path_id, local_root_string)
prepared_by_id[base_path_id] = prepared
prepared_by_old_path.setdefault(old_path, []).append(prepared)
if first_prepared is None:
first_prepared = prepared
item["path"] = local_root_string
for asset in assets_by_owner.get((owner_type, owner_id), []):
relative_path = _relative_asset_path(asset, logical_root)
if not relative_path:
continue
target = _safe_local_target(local_root, relative_path)
target.parent.mkdir(parents=True, exist_ok=True)
version, blob = version_blobs[asset.id]
if include_bytes:
try:
data = backend.get_bytes(blob.storage_key) if backend else b""
except StorageBackendError as exc:
raise FileStorageError(str(exc)) from exc
target.write_bytes(data)
else:
target.touch()
local_key = str(target.resolve())
manifest[local_key] = ManagedAttachmentFile(
local_path=local_key,
asset_id=asset.id,
version_id=version.id,
blob_id=blob.id,
display_path=asset.display_path,
relative_path=normalize_logical_path(relative_path),
filename=asset.filename,
owner_type=asset.owner_type,
owner_id=owner_id,
checksum_sha256=blob.checksum_sha256,
size_bytes=blob.size_bytes,
content_type=blob.content_type,
linked_to_campaign=asset.id in linked_asset_ids,
source_provenance=source_provenance_from_metadata(asset.metadata_),
source_revision=source_revision_from_metadata(asset.metadata_),
) )
for rule in _iter_rule_dicts(attachments, prepared_json):
selected = _selected_base_path(rule, prepared_by_id, prepared_by_old_path, first_prepared)
if selected is None:
continue
base_path_id, local_root_string = selected
rule["base_path_id"] = base_path_id
rule["base_dir"] = local_root_string
if first_prepared is not None:
attachments["base_path"] = first_prepared[1]
snapshot_path = destination / "campaign.json" snapshot_path = destination / "campaign.json"
snapshot_path.write_text(json.dumps(prepared_json, ensure_ascii=False, indent=2), encoding="utf-8") snapshot_path.write_text(json.dumps(prepared_json, ensure_ascii=False, indent=2), encoding="utf-8")