diff --git a/src/albert/__init__.py b/src/albert/__init__.py index 1c3bd16c..5629ed49 100644 --- a/src/albert/__init__.py +++ b/src/albert/__init__.py @@ -4,4 +4,4 @@ __all__ = ["Albert", "AlbertClientCredentials", "AlbertSSOClient"] -__version__ = "1.11.1" +__version__ = "1.11.2" diff --git a/src/albert/collections/worksheets.py b/src/albert/collections/worksheets.py index 6a61660b..d94ab499 100644 --- a/src/albert/collections/worksheets.py +++ b/src/albert/collections/worksheets.py @@ -3,6 +3,7 @@ from albert.collections.base import BaseCollection from albert.core.session import AlbertSession from albert.core.shared.identifiers import ProjectId +from albert.resources.sheets import CellType, Sheet from albert.resources.worksheets import Worksheet @@ -116,3 +117,133 @@ def add_sheet(self, *, project_id: ProjectId, sheet_name: str) -> Worksheet: url = f"{self.base_path}/project/{project_id}/sheets" self.session.put(url=url, json=payload) return self.get_by_project_id(project_id=project_id) + + @validate_call + def duplicate_sheet( + self, + *, + project_id: ProjectId, + source_sheet_name: str, + new_sheet_name: str, + copy_all_pd_rows: bool = True, + copy_all_pinned_columns: bool = True, + copy_all_unpinned_columns: bool = True, + column_names: list[str] | None = None, + task_row_names: list[str] | None = None, + ) -> Worksheet: + """Duplicate an existing sheet within the same project. + + This creates a new sheet based on the specified source sheet. You can control + which Product Design (PD) & Results rows and columns are copied using the available options. + The final list of columns copied is the union of: + - all pinned columns (if copy_all_pinned_columns is True) + - all unpinned columns (if copy_all_unpinned_columns is True) + - explicitly listed column names (column_names) + + Parameters + ---------- + project_id : str + The project ID under which the sheet exists. + source_sheet_name : str + The name of the existing sheet to duplicate. + new_sheet_name : str + The name of the new sheet to create. + copy_all_pd_rows : bool, optional + When True, all PD (Product Design) rows from the source sheet are copied. + When False, only rows corresponding to the selected columns will be copied. + Default is True. + copy_all_pinned_columns : bool, optional + If True, includes all pinned columns from the source sheet. Default is True. + copy_all_unpinned_columns : bool, optional + If True, includes all unpinned columns from the source sheet. Default is True. + column_names : list of str, optional + A list of column names to explicitly copy. These are resolved internally + to column IDs using the sheet's product design grid. + task_row_names : list of str, optional + List of task row names to include from the tasks. + + Returns + ------- + Worksheet + The Worksheet entity containing newly created sheet. + """ + + def _get_sheet_from_worksheet(sheet_name: str, worksheet: Worksheet) -> Sheet: + """Get the sheet from the worksheet by sheet name.""" + sheet = next((s for s in worksheet.sheets if s.name == sheet_name), None) + if not sheet: + raise ValueError(f"Sheet with name {sheet_name} not found in the Worksheet.") + return sheet + + def _get_columns_to_copy(sheet: Sheet, input_column_names: list[str] | None) -> list[str]: + """Determine which columns to copy based on the provided options.""" + sheet_columns = sheet.columns + # Map column names to IDs + all_columns = {col.name: col.column_id for col in sheet_columns} + + # If both flags are true, copy everything + if copy_all_pinned_columns and copy_all_unpinned_columns: + return set(all_columns.values()) + + columns_to_copy: set[str] = set() + # Copy pinned columns + if copy_all_pinned_columns: + columns_to_copy.update( + col.column_id for col in sheet_columns if getattr(col, "pinned", False) + ) + + # Copy unpinned columns + if copy_all_unpinned_columns: + columns_to_copy.update( + col.column_id for col in sheet_columns if not getattr(col, "pinned", False) + ) + + # Add any explicitly specified columns + if input_column_names: + for name in input_column_names: + if name not in all_columns: + raise ValueError(f"Column name {name!r} not found in sheet {sheet.name!r}") + columns_to_copy.add(all_columns[name]) + + return columns_to_copy + + def _get_task_rows_to_copy(sheet: Sheet, input_row_names: list[str] | None) -> list[str]: + """Determine which task rows to copy""" + task_rows = [] + + sheet_rows = sheet.rows + if not input_row_names: + # Copy all task rows if no input rows specified + for row in sheet_rows: + if row.type == CellType.TAS: + task_rows.append(row.row_id) + return task_rows + + name_to_id = {row.name: row.row_id for row in sheet_rows} + for name in input_row_names: + row_id = name_to_id.get(name) + if row_id: + task_rows.append(row_id) + else: + raise ValueError(f"Task row name '{name}' not found in the grid.") + return task_rows + + worksheet = self.get_by_project_id(project_id=project_id) + sheet = _get_sheet_from_worksheet(sheet_name=source_sheet_name, worksheet=worksheet) + columns = _get_columns_to_copy(sheet=sheet, input_column_names=column_names) + task_rows = _get_task_rows_to_copy(sheet=sheet, input_row_names=task_row_names) + + payload = { + "name": new_sheet_name, + "sourceData": { + "projectId": project_id, + "sheetId": sheet.id, + "Columns": [{"id": col_id} for col_id in columns], + "copyAllPDRows": copy_all_pd_rows, + "TaskRows": [{"id": row_id} for row_id in task_rows], + }, + } + + path = f"{self.base_path}/project/{project_id}/sheets" + self.session.put(path, json=payload) + return self.get_by_project_id(project_id=project_id)