Add babs init --shared-group <GROUP> (#368)

Author: tien-tong

babs/base.py

@@ -1,5 +1,6 @@
 """This is the main module."""
 
+import configparser
 import os
 import os.path as op
 import subprocess
@@ -127,6 +128,7 @@ def __init__(self, project_root):
         self.job_status_path_rel = 'code/job_status.csv'
         self.job_status_path_abs = op.join(self.analysis_path, self.job_status_path_rel)
         self.job_submit_path_abs = op.join(self.analysis_path, 'code/job_submit.csv')
+        self._shared_group_enabled_cache = None
         self._apply_config()
 
     def _apply_config(self) -> None:
@@ -177,6 +179,7 @@ def _apply_config(self) -> None:
 
         self.input_datasets = InputDatasets(self.processing_level, config_yaml['input_datasets'])
         self.input_datasets.update_abs_paths(Path(self.project_root) / 'analysis')
+        self.ensure_shared_group_git_safe_directories()
 
     def _validate_pipeline_config(self) -> None:
         """Validate the pipeline configuration if present.
@@ -323,6 +326,118 @@ def wtf_key_info(self, flag_output_ria_only=False) -> None:
                 proc_analysis_dataset_id.stdout.decode('utf-8').strip().lstrip("'").rstrip("'")
             )
 
+    @staticmethod
+    def source_to_local_path(source: str) -> str | None:
+        """Convert a local dataset source URL/path to a filesystem path."""
+        if not source:
+            return None
+        if source.startswith('ria+file://'):
+            local_path = source[len('ria+file://') :]
+        elif source.startswith('file://'):
+            local_path = source[len('file://') :]
+        elif '://' not in source:
+            local_path = source
+        else:
+            return None
+        return local_path.split('#', 1)[0]
+
+    def analysis_git_config_path(self) -> str | None:
+        """Return absolute path to analysis git config file."""
+        git_path = op.join(self.analysis_path, '.git')
+        # Standard repo layout: `.git/` is a directory containing `config`.
+        if op.isdir(git_path):
+            config_path = op.join(git_path, 'config')
+            return config_path if op.exists(config_path) else None
+
+        # Alternate layout (e.g., worktree/submodule): `.git` is a file with
+        # a pointer like `gitdir: /actual/path/to/gitdir`.
+        if op.isfile(git_path):
+            with open(git_path) as f:
+                first_line = f.readline().strip()
+            if first_line.startswith('gitdir:'):
+                git_dir = first_line.split(':', 1)[1].strip()
+                # The gitdir pointer can be relative to `analysis_path`.
+                if not op.isabs(git_dir):
+                    git_dir = op.normpath(op.join(self.analysis_path, git_dir))
+                config_path = op.join(git_dir, 'config')
+                return config_path if op.exists(config_path) else None
+        # No recognizable git metadata/config path found.
+        return None
+
+    def is_shared_group_project(self) -> bool:
+        """Return whether this project is configured for shared-group permissions."""
+        # During `babs init`, shared mode is known from the CLI flag directly.
+        if getattr(self, 'shared_group', None) is not None:
+            return True
+        # Reuse the computed value to avoid repeatedly reading git config.
+        if self._shared_group_enabled_cache is not None:
+            return self._shared_group_enabled_cache
+
+        # For existing projects, infer shared mode from git's core setting.
+        config_path = self.analysis_git_config_path()
+        if config_path is None:
+            self._shared_group_enabled_cache = False
+            return False
+
+        parser = configparser.ConfigParser()
+        parser.read(config_path)
+        # Git may encode shared mode as text ('group') or truthy values.
+        shared_mode = parser.get('core', 'sharedrepository', fallback='').strip().lower()
+        self._shared_group_enabled_cache = shared_mode in {'group', '1', 'true', 'yes'}
+        return self._shared_group_enabled_cache
+
+    def ensure_shared_group_git_safe_directories(self) -> None:
+        """Register project repositories in git safe.directory for shared mode."""
+        if not self.is_shared_group_project():
+            return
+
+        safe_dirs = set()
+        if op.exists(self.analysis_path):
+            safe_dirs.add(op.realpath(self.analysis_path))
+
+        for ria_root in (self.input_ria_path, self.output_ria_path):
+            ria_root_path = Path(ria_root)
+            if not ria_root_path.exists():
+                continue
+            for repo_dir in ria_root_path.glob('*/*'):
+                if repo_dir.is_dir():
+                    safe_dirs.add(str(repo_dir.resolve()))
+
+        alias_data = op.join(self.output_ria_path, 'alias', 'data')
+        if op.exists(alias_data):
+            safe_dirs.add(op.realpath(alias_data))
+
+        if hasattr(self, 'input_datasets'):
+            for in_ds in self.input_datasets:
+                local_path = self.source_to_local_path(getattr(in_ds, 'origin_url', ''))
+                if local_path is not None:
+                    safe_dirs.add(op.realpath(local_path))
+
+        proc_existing = subprocess.run(
+            ['git', 'config', '--global', '--get-all', 'safe.directory'],
+            capture_output=True,
+            text=True,
+        )
+        existing = (
+            {line.strip() for line in proc_existing.stdout.splitlines() if line.strip()}
+            if proc_existing.returncode == 0
+            else set()
+        )
+
+        for repo_path in sorted(safe_dirs):
+            if not op.exists(repo_path) or repo_path in existing:
+                continue
+            subprocess.run(
+                ['git', 'config', '--global', '--add', 'safe.directory', repo_path],
+                capture_output=True,
+                text=True,
+            )
+            existing.add(repo_path)
+
+    def ensure_shared_group_runtime_ready(self) -> None:
+        """Apply git-safe-directory safeguards for shared projects."""
+        self.ensure_shared_group_git_safe_directories()
+
     @property
     def analysis_datalad_handle(self) -> dlapi.Dataset:
         """Cached property of `analysis_datalad_handle`."""
@@ -444,12 +559,24 @@ def get_latest_submitted_jobs_df(self):
         1  sub-0002       1        2
 
         """
+        expected_columns = get_latest_submitted_jobs_columns(self.processing_level)
         if not op.exists(self.job_submit_path_abs):
-            return EMPTY_JOB_STATUS_DF
-        df = pd.read_csv(self.job_submit_path_abs)
-        for column_name in get_latest_submitted_jobs_columns(self.processing_level):
+            return pd.DataFrame(columns=expected_columns)
+
+        try:
+            df = pd.read_csv(self.job_submit_path_abs)
+        except pd.errors.EmptyDataError:
+            return pd.DataFrame(columns=expected_columns)
+
+        # job_submit.csv is written once before submit (without job_id) and then
+        # rewritten after submit (with job_id). Tolerate the interim schema so
+        # follow-up commands do not crash if submit was interrupted.
+        for column_name in expected_columns:
+            if column_name not in df.columns:
+                df[column_name] = pd.NA
             df[column_name] = df[column_name].astype(status_dtypes[column_name])
-        return df
+
+        return df[expected_columns]
 
     def get_currently_running_jobs_df(self):
         """

babs/bootstrap.py

@@ -37,6 +37,7 @@ def babs_bootstrap(
         container_config,
         initial_inclusion_df=None,
         throttle=None,
+        shared_group=None,
     ):
         """
         Bootstrap a babs project: initialize datalad-tracked RIAs, generate scripts to be used, etc
@@ -62,6 +63,10 @@ def babs_bootstrap(
             simultaneously running array tasks. The value will be added to the array
             specification as `%<throttle>`. Example: `10` will result in
             `--array=1-${max_array}%10`.
+        shared_group: str or None, optional
+            Unix group name for shared write access. If provided, `analysis` is
+            initialized with `git init --shared=group` and RIA siblings are created
+            with `--shared group --group <GROUP>`.
         """
         if op.exists(self.project_root):
             raise FileExistsError(
@@ -89,6 +94,7 @@ def babs_bootstrap(
 
         # Store throttle value for job submission template
         self.throttle = throttle
+        self.shared_group = shared_group
 
         # validate `processing_level`:
         self.processing_level = validate_processing_level(processing_level)
@@ -111,9 +117,10 @@ def babs_bootstrap(
         # Create `analysis` folder: -----------------------------
         print('DataLad version: ' + get_datalad_version())
         print('\nCreating `analysis` folder (also a datalad dataset)...')
-        self._analysis_datalad_handle = dlapi.create(
-            self.analysis_path, cfg_proc='yoda', annex=True
-        )
+        create_kwargs = {'cfg_proc': 'yoda', 'annex': True}
+        if self.shared_group is not None:
+            create_kwargs['initopts'] = ['--shared=group']
+        self._analysis_datalad_handle = dlapi.create(self.analysis_path, **create_kwargs)
         self.input_datasets.update_abs_paths(Path(self.analysis_path))
         self.input_datasets.set_inclusion_dataframe(initial_inclusion_df, processing_level)
 
@@ -171,8 +178,14 @@ def babs_bootstrap(
         )
         # Create output RIA sibling: -----------------------------
         print('\nCreating output and input RIA...')
+        sibling_kwargs = {}
+        if self.shared_group is not None:
+            sibling_kwargs = {'shared': 'group', 'group': self.shared_group}
         self.analysis_datalad_handle.create_sibling_ria(
-            name='output', url=self.output_ria_url, new_store_ok=True
+            name='output',
+            url=self.output_ria_url,
+            new_store_ok=True,
+            **sibling_kwargs,
         )
 
         self.wtf_key_info()
@@ -183,6 +196,7 @@ def babs_bootstrap(
             url=self.input_ria_url,
             storage_sibling=False,  # False is `off` in CLI of datalad
             new_store_ok=True,
+            **sibling_kwargs,
         )
 
         # Register the input dataset(s): -----------------------------
@@ -383,6 +397,7 @@ def babs_bootstrap(
 
         # Initialize the job status csv file:
         self._create_initial_job_status_csv()
+        self.ensure_shared_group_git_safe_directories()
 
         print('\n')
         print(
@@ -403,7 +418,13 @@ def _bootstrap_single_app_scripts(
         print('\nGenerating a bash script for running container and zipping the outputs...')
         print('This bash script will be named as `' + container_name + '_zip.sh`')
         bash_path = op.join(self.analysis_path, 'code', container_name + '_zip.sh')
-        container.generate_bash_run_bidsapp(bash_path, self.input_datasets, self.processing_level)
+        shared_group_mode = self.shared_group is not None
+        container.generate_bash_run_bidsapp(
+            bash_path,
+            self.input_datasets,
+            self.processing_level,
+            shared_group_mode=shared_group_mode,
+        )
         self.datalad_save(
             path='code/' + container_name + '_zip.sh',
             message='Generate script of running container',
@@ -422,11 +443,16 @@ def _bootstrap_single_app_scripts(
             self.processing_level,
             system,
             project_root=op.dirname(self.analysis_path),
+            shared_group_mode=shared_group_mode,
         )
 
         # also, generate a bash script of a test job used by `babs check-setup`:
         path_check_setup = op.join(self.analysis_path, 'code/check_setup')
-        container.generate_bash_test_job(path_check_setup, system)
+        container.generate_bash_test_job(
+            path_check_setup,
+            system,
+            shared_group_mode=shared_group_mode,
+        )
 
     def _bootstrap_pipeline_scripts(self, container_ds, container_config, system):
         """Bootstrap scripts for pipeline configuration."""
@@ -457,7 +483,7 @@ def _bootstrap_pipeline_scripts(self, container_ds, container_config, system):
 
         with open(pipeline_script_path, 'w') as f:
             f.write(pipeline_script_content)
-        os.chmod(pipeline_script_path, 0o700)
+        os.chmod(pipeline_script_path, 0o770 if self.shared_group is not None else 0o700)
 
         self.datalad_save(
             path='code/pipeline_zip.sh',
@@ -496,7 +522,7 @@ def _bootstrap_pipeline_scripts(self, container_ds, container_config, system):
 
         with open(bash_path, 'w') as f:
             f.write(participant_job_content)
-        os.chmod(bash_path, 0o700)
+        os.chmod(bash_path, 0o770 if self.shared_group is not None else 0o700)
 
         # also, generate bash scripts of test jobs used by `babs check-setup`:
         # Generate test jobs for all containers in the pipeline
@@ -511,7 +537,11 @@ def _bootstrap_pipeline_scripts(self, container_ds, container_config, system):
             # Create separate test job directories for each container
             step_check_setup = op.join(path_check_setup, f'step_{i + 1}_{step_container_name}')
             os.makedirs(step_check_setup, exist_ok=True)
-            step_container.generate_bash_test_job(step_check_setup, system)
+            step_container.generate_bash_test_job(
+                step_check_setup,
+                system,
+                shared_group_mode=(self.shared_group is not None),
+            )
 
     def _init_import_files(self, file_list):
         """

babs/cli.py

@@ -126,6 +126,14 @@ def _parse_init():
         'The value will be added to the array specification as ``%%<throttle>``. '
         'Example: ``--throttle 10`` will result in ``--array=1-${max_array}%%10``.',
     )
+    parser.add_argument(
+        '--shared_group',
+        '--shared-group',
+        type=str,
+        help='Unix group name for shared write access. If provided, '
+        '`analysis` is initialized with ``git init --shared=group`` and '
+        'RIA siblings are created with ``--shared group --group <GROUP>``.',
+    )
 
     return parser
 
@@ -157,6 +165,7 @@ def babs_init_main(
     queue: str,
     keep_if_failed: bool,
     throttle: int | None = None,
+    shared_group: str | None = None,
 ):
     """This is the core function of babs init.
 
@@ -189,6 +198,10 @@ def babs_init_main(
         simultaneously running array tasks. The value will be added to the array
         specification as `%<throttle>`. Example: `10` will result in
         `--array=1-${max_array}%10`.
+    shared_group: str or None, optional
+        Unix group name for shared write access. If provided, `analysis` is
+        initialized with `git init --shared=group` and RIA siblings are created
+        with `--shared group --group <GROUP>`.
     """
 
     from babs import BABSBootstrap
@@ -203,6 +216,7 @@ def babs_init_main(
             container_config,
             list_sub_file,
             throttle=throttle,
+            shared_group=shared_group,
         )
     except Exception as exc:
         print('\n`babs init` failed! Below is the error message:')