feat: allow definition of access pattern for input files

snakemake/access_patterns.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+from typing import Optional
+from snakemake.io import flag
+from snakemake_interface_common.exceptions import WorkflowError
+
+VALID_PATTERNS = ["head", "tail", "random", "sequential"]
+
+
+@dataclass
+class AccessPattern:
+    pattern: str
+    lines: Optional[int] = None
+    bytes: Optional[int] = None
+
+
+def access_pattern(
+    path, pattern: str, lines: Optional[int] = None, bytes: Optional[int] = None
+):
+    if pattern not in VALID_PATTERNS:
+        raise WorkflowError(
+            f"Invalid access pattern: {pattern}. Valid patterns are: {', '.join(VALID_PATTERNS)}"
+        )
+    return flag(path, "access_pattern", AccessPattern(pattern, lines, bytes))

snakemake/workflow.py

@@ -84,6 +84,7 @@
     IOFile,
     sourcecache_entry,
 )
+from snakemake.access_patterns import access_pattern
 
 from snakemake.persistence import Persistence
 from snakemake.utils import update_config
@@ -205,6 +206,7 @@ def __post_init__(self):
         _globals["gitlab"] = sourcecache.GitlabFile
         _globals["gitfile"] = sourcecache.LocalGitFile
         _globals["storage"] = self._storage_registry
+        _globals["access_pattern"] = access_pattern
 
         self.vanilla_globals = dict(_globals)
         self.modifier_stack = [WorkflowModifier(self, globals=_globals)]