Task Collections

class Collection:
    """
    Hierarchical container for tasks and sub-collections.
    
    Provides namespace management, task organization, auto-discovery from
    modules, and configuration integration for building complex CLI applications.
    
    Attributes:
    - tasks (dict): Mapping of task names to Task objects
    - collections (dict): Mapping of collection names to Collection objects  
    - default (str): Name of default task
    - name (str): Collection name
    - loaded_from (str): Path collection was loaded from
    - auto_dash_names (bool): Convert underscores to dashes in task names
    - _configuration (dict): Collection-specific configuration
    """
    
    def __init__(self, *args, **kwargs):
        """
        Initialize Collection.
        
        Parameters:
        - *args: Tasks or collections to add
        - name (str, optional): Collection name
        - default (str, optional): Default task name
        - auto_dash_names (bool): Convert underscores to dashes
        - **kwargs: Additional task name -> task mappings
        """
    
    def add_task(self, task, name=None, aliases=None, default=None):
        """
        Add task to collection.
        
        Parameters:
        - task (Task or callable): Task to add
        - name (str, optional): Task name (defaults to task.name or function name)
        - aliases (list, optional): Alternative names for task
        - default (bool, optional): Whether this is the default task
        
        Returns:
        Task: Added task object
        """
    
    def add_collection(self, coll, name=None):
        """
        Add sub-collection to this collection.
        
        Parameters:
        - coll (Collection): Collection to add
        - name (str, optional): Name for sub-collection
        """
    
    @classmethod
    def from_module(cls, module, name=None, config=None, loaded_from=None, auto_dash_names=True):
        """
        Create collection from Python module.
        
        Automatically discovers tasks (decorated functions) and sub-collections
        in the given module and creates a Collection containing them.
        
        Parameters:
        - module: Python module to load from
        - name (str, optional): Collection name
        - config (Config, optional): Configuration object
        - loaded_from (str, optional): Module file path
        - auto_dash_names (bool): Convert underscores to dashes
        
        Returns:
        Collection: Collection loaded from module
        """
    
    def subcollection_from_path(self, path):
        """
        Get sub-collection from dotted path.
        
        Parameters:
        - path (str): Dotted path to sub-collection (e.g., 'deploy.staging')
        
        Returns:
        Collection: Sub-collection at path
        
        Raises:
        CollectionNotFound: If path not found
        """
    
    def task_with_config(self, name, config):
        """
        Get task with merged configuration.
        
        Parameters:
        - name (str): Task name
        - config (Config): Base configuration
        
        Returns:
        Task: Task with merged configuration
        """
    
    def to_contexts(self, config):
        """
        Convert collection to parser contexts.
        
        Parameters:
        - config (Config): Configuration object
        
        Returns:
        list: List of ParserContext objects for CLI parsing
        """
    
    def transform(self, name, config):
        """
        Transform task name according to collection settings.
        
        Parameters:
        - name (str): Original task name
        - config (Config): Configuration object
        
        Returns:
        str: Transformed task name
        """
    
    def configuration(self, taskname=None):
        """
        Get configuration for task or collection.
        
        Parameters:
        - taskname (str, optional): Specific task name
        
        Returns:
        dict: Configuration dictionary
        """
    
    def configure(self, options):
        """
        Configure collection with options.
        
        Parameters:
        - options (dict): Configuration options
        """
    
    def serialized(self):
        """
        Get serialized representation of collection.
        
        Returns:
        dict: Serialized collection data
        """
    
    @property
    def task_names(self):
        """
        Get list of all task names in collection.
        
        Returns:
        list: Task names
        """
    
    def __getitem__(self, name):
        """Get task or sub-collection by name."""
    
    def __contains__(self, name):
        """Test if task or collection exists."""
    
    def __iter__(self):
        """Iterate over task and collection names."""

class FilesystemLoader:
    """
    Loads Collections from filesystem.
    
    Handles loading task collections from Python modules and packages
    on the filesystem with support for nested discovery and import handling.
    
    Attributes:
    - start (str): Starting directory for collection discovery
    """
    
    def __init__(self, start=None):
        """
        Initialize FilesystemLoader.
        
        Parameters:
        - start (str, optional): Starting directory path
        """
    
    def find(self, name):
        """
        Find module for given collection name.
        
        Parameters:
        - name (str): Collection name to find
        
        Returns:
        str: Path to module file
        
        Raises:
        CollectionNotFound: If collection not found
        """
    
    def load(self, name=None):
        """
        Load collection by name.
        
        Parameters:
        - name (str, optional): Collection name to load
        
        Returns:
        Collection: Loaded collection
        
        Raises:
        CollectionNotFound: If collection not found
        """
    
    @property
    def start(self):
        """Starting directory for collection discovery."""

class Executor:
    """
    Task execution orchestrator.
    
    Handles task execution order, dependency resolution, call deduplication,
    and execution coordination for complex task workflows.
    
    Attributes:
    - collection (Collection): Task collection
    - config (Config): Configuration object
    - core (list): Core argument list
    """
    
    def __init__(self, collection, config=None, core=None):
        """
        Initialize Executor.
        
        Parameters:
        - collection (Collection): Task collection
        - config (Config, optional): Configuration object
        - core (list, optional): Core arguments
        """
    
    def execute(self, *tasks):
        """
        Execute tasks in proper order.
        
        Parameters:
        - *tasks: Task names or Call objects to execute
        
        Returns:
        dict: Execution results
        """
    
    def normalize(self, tasks):
        """
        Normalize task specifications to Call objects.
        
        Parameters:
        - tasks (list): Task specifications
        
        Returns:
        list: Normalized Call objects
        """
    
    def dedupe(self, tasks):
        """
        Remove duplicate task calls.
        
        Parameters:
        - tasks (list): List of Call objects
        
        Returns:
        list: Deduplicated Call objects
        """
    
    def expand_calls(self, calls):
        """
        Expand calls with pre/post task dependencies.
        
        Parameters:
        - calls (list): Call objects to expand
        
        Returns:
        list: Expanded calls with dependencies
        """

from invoke import Collection, task

@task
def hello(ctx, name='World'):
    """Say hello."""
    print(f"Hello {name}!")

@task
def goodbye(ctx, name='World'):
    """Say goodbye."""
    print(f"Goodbye {name}!")

# Create collection manually
ns = Collection()
ns.add_task(hello)
ns.add_task(goodbye, default=True)  # Make goodbye the default task

# Or create from tasks directly
ns = Collection(hello, goodbye)

# Or use keyword arguments
ns = Collection(greet=hello, farewell=goodbye)

# tasks.py
from invoke import task, Collection

@task
def clean(ctx):
    """Clean build artifacts."""
    ctx.run("rm -rf build/ dist/ *.egg-info/")

@task
def build(ctx):
    """Build the project.""" 
    ctx.run("python setup.py build")

@task
def test(ctx):
    """Run tests."""
    ctx.run("python -m pytest")

# Optional: Create sub-collections
deploy = Collection('deploy')

@task
def staging(ctx):
    """Deploy to staging."""
    ctx.run("deploy-staging.sh")

@task  
def production(ctx):
    """Deploy to production."""
    ctx.run("deploy-production.sh")

deploy.add_task(staging)
deploy.add_task(production)

# Main collection includes deploy sub-collection
ns = Collection(clean, build, test, deploy)

from invoke import Collection

# Load collection from tasks.py module
import tasks
ns = Collection.from_module(tasks)

# Or load by module name
ns = Collection.from_module('tasks')

from invoke import Collection, task

# Database tasks 
@task
def migrate(ctx):
    """Run database migrations."""
    ctx.run("python manage.py migrate")

@task
def seed(ctx):
    """Seed database with test data."""
    ctx.run("python manage.py loaddata fixtures.json")

db = Collection('db')
db.add_task(migrate)
db.add_task(seed)

# Deployment tasks
@task
def staging(ctx):
    """Deploy to staging."""
    ctx.run("deploy-staging.sh")

@task
def production(ctx):
    """Deploy to production."""
    ctx.run("deploy-production.sh")

deploy = Collection('deploy')
deploy.add_task(staging)
deploy.add_task(production)

# Main collection
ns = Collection(db, deploy)

# Usage:
# invoke db.migrate
# invoke db.seed  
# invoke deploy.staging
# invoke deploy.production

from invoke import Collection, task, Config

@task
def serve(ctx):
    """Start development server."""
    port = ctx.config.server.port
    host = ctx.config.server.host
    ctx.run(f"python manage.py runserver {host}:{port}")

# Create collection with configuration
config = Config({
    'server': {
        'host': 'localhost',
        'port': 8000
    }
})

ns = Collection(serve)
ns.configure({'server': {'port': 3000}})  # Override port

# Directory structure:
# myproject/
#   tasks/
#     __init__.py
#     build.py  
#     deploy.py
#     test.py

# tasks/__init__.py
from invoke import Collection
from . import build, deploy, test

ns = Collection.from_module(build, name='build')
ns.add_collection(Collection.from_module(deploy, name='deploy'))
ns.add_collection(Collection.from_module(test, name='test'))

# tasks/build.py
from invoke import task

@task
def clean(ctx):
    """Clean build artifacts."""
    ctx.run("rm -rf build/")

@task  
def compile(ctx):
    """Compile source code."""
    ctx.run("python setup.py build")

# Usage:
# invoke build.clean
# invoke build.compile
# invoke deploy.staging
# invoke test.unit

from invoke import Collection, task

@task
def clean(ctx):
    """Clean build directory."""
    ctx.run("rm -rf build/")

@task
def compile(ctx):
    """Compile source code."""
    ctx.run("gcc -o myapp src/*.c")

@task(pre=[clean, compile])
def package(ctx):
    """Package application."""
    ctx.run("tar -czf myapp.tar.gz myapp")

@task(pre=[package])
def deploy(ctx):
    """Deploy packaged application."""
    ctx.run("scp myapp.tar.gz server:/opt/")

ns = Collection(clean, compile, package, deploy)

# Running 'invoke deploy' will automatically run:
# 1. clean
# 2. compile  
# 3. package
# 4. deploy

from invoke import Collection, task

@task
def hello(ctx):
    pass

@task  
def world(ctx):
    pass

ns = Collection(hello, world)

# List all tasks
print(ns.task_names)  # ['hello', 'world']

# Check if task exists
print('hello' in ns)  # True

# Get task object
task_obj = ns['hello']
print(task_obj.name)  # 'hello'

# Iterate over tasks
for name in ns:
    task_obj = ns[name]
    print(f"Task: {name}")

from invoke.loader import FilesystemLoader

# Create loader
loader = FilesystemLoader()

# Find task module
module_path = loader.find('tasks')  # Looks for tasks.py or tasks/__init__.py

# Load collection
collection = loader.load('tasks')

# Use with custom start directory
loader = FilesystemLoader(start='/path/to/project')
collection = loader.load('deployment_tasks')