chore: sync GovOPlaN module split state
This commit is contained in:
@@ -3,8 +3,9 @@ from __future__ import annotations
|
||||
import mimetypes
|
||||
import zipfile
|
||||
from io import BytesIO
|
||||
from os import PathLike
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
from typing import Any, Iterable
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
@@ -68,11 +69,12 @@ def extract_zip_upload(
|
||||
owner_type: str,
|
||||
owner_id: str,
|
||||
user_id: str,
|
||||
zip_data: bytes,
|
||||
zip_data: bytes | str | PathLike[str],
|
||||
folder: str | None,
|
||||
campaign_id: str | None,
|
||||
conflict_strategy: str = "reject",
|
||||
conflict_resolutions: Iterable[FileConflictResolution] | None = None,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
is_admin: bool = False,
|
||||
max_files: int = 1000,
|
||||
max_file_bytes: int = 50 * 1024 * 1024,
|
||||
@@ -82,7 +84,8 @@ def extract_zip_upload(
|
||||
total = 0
|
||||
base_folder = normalize_folder(folder)
|
||||
try:
|
||||
with zipfile.ZipFile(BytesIO(zip_data)) as archive:
|
||||
source = BytesIO(zip_data) if isinstance(zip_data, bytes) else zip_data
|
||||
with zipfile.ZipFile(source) as archive:
|
||||
infos = [info for info in archive.infolist() if not info.is_dir()]
|
||||
if len(infos) > max_files:
|
||||
raise FileStorageError(f"ZIP contains too many files (limit {max_files})")
|
||||
@@ -115,6 +118,7 @@ def extract_zip_upload(
|
||||
data=data,
|
||||
display_path=target_path,
|
||||
content_type=mimetypes.guess_type(inner_path)[0] or "application/octet-stream",
|
||||
metadata=metadata,
|
||||
campaign_id=campaign_id,
|
||||
conflict_strategy=conflict_strategy,
|
||||
conflict_resolutions=conflict_resolutions,
|
||||
|
||||
Reference in New Issue
Block a user