angreal.integrations.docker.compose

The Docker Compose integration provides a Pythonic interface to manage multi-container Docker applications using Docker Compose. It uses subprocess execution to ensure full compatibility with all Docker Compose features and versions.

Create a DockerCompose instance for managing a compose stack.

compose(file: str, project_name: Optional[str] = None) -> DockerCompose

Parameters:

• file (str): Path to the docker-compose.yml file
• project_name (Optional[str]): Override the default project name

Returns:

• DockerCompose: An instance for managing the compose stack

Example:

from angreal.integrations.docker import compose

# Basic usage
stack = compose("docker-compose.yml")

# With custom project name
stack = compose("docker-compose.dev.yml", project_name="myapp-dev")

Manages Docker Compose operations for a specific compose file.

• compose_file (str): Path to the compose file
• working_dir (str): Working directory for compose operations
• project_name (Optional[str]): Project name if specified

Check if Docker Compose is available on the system.

@staticmethod
is_available() -> bool

Returns:

• bool: True if Docker Compose (v1 or v2) is available

Example:

if not DockerCompose.is_available():
    print("Docker Compose is not installed")
    return

Start services defined in the compose file.

up(detach: bool = True,
   build: bool = False,
   remove_orphans: bool = False,
   force_recreate: bool = False,
   no_recreate: bool = False,
   services: Optional[List[str]] = None) -> ComposeResult

Parameters:

• detach (bool): Run containers in the background (default: True)
• build (bool): Build images before starting containers (default: False)
• remove_orphans (bool): Remove containers for services not in compose file (default: False)
• force_recreate (bool): Recreate containers even if configuration unchanged (default: False)
• no_recreate (bool): Don’t recreate containers if they exist (default: False)
• services (Optional[List[str]]): Specific services to start (default: all services)

Returns:

• ComposeResult: Operation result with success status and output

Example:

# Start all services in detached mode
result = stack.up(detach=True)

# Build and start specific services
result = stack.up(build=True, services=["web", "db"])

# Force recreate all containers
result = stack.up(force_recreate=True, remove_orphans=True)

Stop and remove containers, networks, and optionally volumes.

down(volumes: bool = False,
     remove_orphans: bool = False,
     timeout: Optional[str] = None) -> ComposeResult

Parameters:

• volumes (bool): Remove named volumes (default: False)
• remove_orphans (bool): Remove containers for services not in compose file (default: False)
• timeout (Optional[str]): Shutdown timeout in seconds (default: 10)

Returns:

• ComposeResult: Operation result

Example:

# Stop and remove containers
result = stack.down()

# Also remove volumes
result = stack.down(volumes=True)

# Custom timeout
result = stack.down(timeout="30")

Restart services.

restart(services: Optional[List[str]] = None,
        timeout: Optional[str] = None) -> ComposeResult

Parameters:

• services (Optional[List[str]]): Specific services to restart (default: all)
• timeout (Optional[str]): Shutdown timeout in seconds

Returns:

• ComposeResult: Operation result

Example:

# Restart all services
result = stack.restart()

# Restart specific service
result = stack.restart(services=["web"])

View output from services.

logs(services: Optional[List[str]] = None,
     follow: bool = False,
     timestamps: bool = False,
     tail: Optional[str] = None,
     since: Optional[str] = None) -> ComposeResult

Parameters:

• services (Optional[List[str]]): Specific services (default: all)
• follow (bool): Follow log output (default: False)
• timestamps (bool): Show timestamps (default: False)
• tail (Optional[str]): Number of lines to show from end
• since (Optional[str]): Show logs since timestamp or relative time

Returns:

• ComposeResult: Operation result with logs in stdout

Example:

# Get last 100 lines from web service
result = stack.logs(services=["web"], tail="100")

# Follow logs with timestamps
result = stack.logs(follow=True, timestamps=True)

# Logs from last hour
result = stack.logs(since="1h")

List containers.

ps(all: bool = False,
   quiet: bool = False,
   services: bool = False,
   filter_services: Optional[List[str]] = None) -> ComposeResult

Parameters:

• all (bool): Show all containers including stopped (default: False)
• quiet (bool): Only display container IDs (default: False)
• services (bool): Display services (default: False)
• filter_services (Optional[List[str]]): Filter by service names

Returns:

• ComposeResult: Operation result with container list

Example:

# List running containers
result = stack.ps()

# Show all containers
result = stack.ps(all=True)

# List services only
result = stack.ps(services=True)

Build or rebuild services.

build(services: Optional[List[str]] = None,
      no_cache: bool = False,
      pull: bool = False,
      parallel: bool = False) -> ComposeResult

Parameters:

• services (Optional[List[str]]): Specific services to build (default: all)
• no_cache (bool): Don’t use cache (default: False)
• pull (bool): Always pull newer base images (default: False)
• parallel (bool): Build in parallel (default: False)

Returns:

• ComposeResult: Operation result

Example:

# Build all services
result = stack.build()

# Force rebuild without cache
result = stack.build(no_cache=True, pull=True)

# Build specific service
result = stack.build(services=["web"])

Start existing containers.

start(services: Optional[List[str]] = None) -> ComposeResult

Parameters:

• services (Optional[List[str]]): Specific services to start

Returns:

• ComposeResult: Operation result

Stop running containers without removing them.

stop(services: Optional[List[str]] = None,
     timeout: Optional[str] = None) -> ComposeResult

Parameters:

• services (Optional[List[str]]): Specific services to stop
• timeout (Optional[str]): Shutdown timeout in seconds

Returns:

• ComposeResult: Operation result

Execute a command in a running container.

exec(service: str,
     command: List[str],
     detach: bool = False,
     tty: bool = True,
     user: Optional[str] = None,
     workdir: Optional[str] = None,
     env: Optional[Dict[str, str]] = None) -> ComposeResult

Parameters:

• service (str): Service name
• command (List[str]): Command to execute
• detach (bool): Detached mode (default: False)
• tty (bool): Allocate a pseudo-TTY (default: True)
• user (Optional[str]): User to run command as
• workdir (Optional[str]): Working directory inside container
• env (Optional[Dict[str, str]]): Environment variables

Returns:

• ComposeResult: Operation result

Example:

# Run bash in web container
result = stack.exec("web", ["bash"])

# Run command as specific user
result = stack.exec("db", ["psql", "-U", "postgres"], user="postgres")

# Run with environment variables
result = stack.exec("app", ["python", "script.py"], env={"DEBUG": "1"})

Pull service images.

pull(services: Optional[List[str]] = None) -> ComposeResult

Parameters:

• services (Optional[List[str]]): Specific services (default: all)

Returns:

• ComposeResult: Operation result

Validate and view the compose configuration.

config(quiet: bool = False,
       services: bool = False,
       volumes: bool = False) -> ComposeResult

Parameters:

• quiet (bool): Only validate configuration (default: False)
• services (bool): Print service names (default: False)
• volumes (bool): Print volume names (default: False)

Returns:

• ComposeResult: Operation result with configuration

Result object returned by all Docker Compose operations.

• success (bool): Whether the operation succeeded
• exit_code (int): Process exit code
• stdout (str): Standard output from the command
• stderr (str): Standard error output from the command

Example:

result = stack.up(detach=True)
if result.success:
    print("Stack started successfully")
else:
    print(f"Failed with exit code {result.exit_code}")
    print(f"Error: {result.stderr}")

from angreal.integrations.docker import compose

# Create and manage a development stack
def manage_dev_stack():
    # Check Docker Compose availability
    if not DockerCompose.is_available():
        print("Docker Compose is not available")
        return 1

    # Initialize stack
    stack = compose("docker-compose.dev.yml", project_name="myapp-dev")

    # Pull latest images
    print("Pulling latest images...")
    result = stack.pull()
    if not result.success:
        print(f"Pull failed: {result.stderr}")
        return 1

    # Build and start services
    print("Starting services...")
    result = stack.up(detach=True, build=True, remove_orphans=True)
    if not result.success:
        print(f"Failed to start: {result.stderr}")
        return 1

    # Check status
    result = stack.ps()
    print("Running services:")
    print(result.stdout)

    # View logs
    result = stack.logs(services=["web"], tail="20")
    print("Recent web logs:")
    print(result.stdout)

    # Execute database migration
    print("Running migrations...")
    result = stack.exec("web", ["python", "manage.py", "migrate"])
    if not result.success:
        print(f"Migration failed: {result.stderr}")

    return 0

All methods return a ComposeResult object. Check the success attribute to determine if the operation succeeded:

result = stack.up(detach=True)
if not result.success:
    # Handle error
    print(f"Error: {result.stderr}")
    print(f"Exit code: {result.exit_code}")

Common error scenarios:

• Docker Compose not installed: Check with DockerCompose.is_available()
• Invalid compose file: Constructor will raise IOError
• Service not found: Check stderr for “no such service”
• Container not running: Exec commands will fail with appropriate message

The integration automatically detects and uses the appropriate Docker Compose version:

1. Tries docker compose (v2) first
2. Falls back to docker-compose (v1) if v2 not available
3. All features work with both versions