Docker Automation with Python

1import docker
2from docker.models.containers import Container
3from docker.models.images import Image
4from docker.types import Mount, LogConfig
5from typing import List, Dict, Optional, Any
6import logging
7from datetime import datetime
8
9logging.basicConfig(level=logging.INFO)
10logger = logging.getLogger(__name__)
11
12
13class DockerContainerManager:
14    """
15    Manage Docker containers with complete lifecycle automation
16    """
17
18    def __init__(self, base_url: str = 'unix://var/run/docker.sock'):
19        """
20        Initialize Docker client
21
22        Args:
23            base_url: Docker daemon socket URL
24        """
25        try:
26            self.client = docker.DockerClient(base_url=base_url)
27            self.api_client = docker.APIClient(base_url=base_url)
28
29            # Verify connection
30            self.client.ping()
31            logger.info("✓ Connected to Docker daemon")
32
33            # Get Docker info
34            info = self.client.info()
35            logger.info(f"  Docker version: {info['ServerVersion']}")
36            logger.info(f"  Containers: {info['Containers']} ({info['ContainersRunning']} running)")
37
38        except docker.errors.DockerException as e:
39            logger.error(f"Failed to connect to Docker: {e}")
40            raise
41
42    def run_container(
43        self,
44        image: str,
45        name: Optional[str] = None,
46        command: Optional[str] = None,
47        environment: Optional[Dict[str, str]] = None,
48        ports: Optional[Dict[str, int]] = None,
49        volumes: Optional[Dict[str, Dict[str, str]]] = None,
50        detach: bool = True,
51        remove: bool = False,
52        network: Optional[str] = None,
53        cpu_limit: Optional[float] = None,
54        mem_limit: Optional[str] = None
55    ) -> Container:
56        """
57        Run a Docker container with comprehensive configuration
58
59        Args:
60            image: Docker image name
61            name: Container name
62            command: Command to execute
63            environment: Environment variables
64            ports: Port mappings {container_port: host_port}
65            volumes: Volume mounts
66            detach: Run in background
67            remove: Remove container when stopped
68            network: Network to attach
69            cpu_limit: CPU limit (number of CPUs)
70            mem_limit: Memory limit (e.g., '512m', '2g')
71
72        Returns:
73            Container instance
74        """
75        logger.info(f"Starting container from image: {image}")
76
77        try:
78            # Ensure image exists
79            try:
80                self.client.images.get(image)
81            except docker.errors.ImageNotFound:
82                logger.info(f"  Image not found locally, pulling: {image}")
83                self.client.images.pull(image)
84
85            # Configure resource limits
86            host_config = {}
87            if cpu_limit:
88                host_config['nano_cpus'] = int(cpu_limit * 1e9)
89            if mem_limit:
90                host_config['mem_limit'] = mem_limit
91
92            # Run container
93            container = self.client.containers.run(
94                image=image,
95                name=name,
96                command=command,
97                environment=environment or {},
98                ports=ports or {},
99                volumes=volumes or {},
100                detach=detach,
101                remove=remove,
102                network=network,
103                **host_config
104            )
105
106            logger.info(f"✓ Container started: {container.short_id}")
107            logger.info(f"  Name: {container.name}")
108            logger.info(f"  Status: {container.status}")
109
110            return container
111
112        except docker.errors.ContainerError as e:
113            logger.error(f"Container execution failed: {e}")
114            raise
115        except docker.errors.ImageNotFound as e:
116            logger.error(f"Image not found: {e}")
117            raise
118        except docker.errors.APIError as e:
119            logger.error(f"Docker API error: {e}")
120            raise
121
122    def list_containers(
123        self,
124        all: bool = False,
125        filters: Optional[Dict] = None
126    ) -> List[Container]:
127        """
128        List Docker containers
129
130        Args:
131            all: Include stopped containers
132            filters: Filter results (e.g., {'status': 'running'})
133
134        Returns:
135            List of Container instances
136        """
137        containers = self.client.containers.list(all=all, filters=filters)
138
139        logger.info(f"Found {len(containers)} containers")
140
141        for container in containers:
142            logger.info(f"  {container.short_id}: {container.name} - {container.status}")
143
144        return containers
145
146    def stop_container(
147        self,
148        container_id: str,
149        timeout: int = 10
150    ) -> bool:
151        """
152        Stop a running container
153
154        Args:
155            container_id: Container ID or name
156            timeout: Seconds to wait before killing
157
158        Returns:
159            True if stopped successfully
160        """
161        try:
162            container = self.client.containers.get(container_id)
163
164            logger.info(f"Stopping container: {container.name}")
165
166            container.stop(timeout=timeout)
167
168            logger.info(f"✓ Container stopped: {container.short_id}")
169
170            return True
171
172        except docker.errors.NotFound:
173            logger.error(f"Container not found: {container_id}")
174            return False
175        except docker.errors.APIError as e:
176            logger.error(f"Failed to stop container: {e}")
177            return False
178
179    def remove_container(
180        self,
181        container_id: str,
182        force: bool = False,
183        volumes: bool = False
184    ) -> bool:
185        """
186        Remove a container
187
188        Args:
189            container_id: Container ID or name
190            force: Force removal of running container
191            volumes: Remove associated volumes
192
193        Returns:
194            True if removed successfully
195        """
196        try:
197            container = self.client.containers.get(container_id)
198
199            logger.info(f"Removing container: {container.name}")
200
201            container.remove(force=force, v=volumes)
202
203            logger.info(f"✓ Container removed: {container_id}")
204
205            return True
206
207        except docker.errors.NotFound:
208            logger.error(f"Container not found: {container_id}")
209            return False
210        except docker.errors.APIError as e:
211            logger.error(f"Failed to remove container: {e}")
212            return False
213
214    def get_container_logs(
215        self,
216        container_id: str,
217        tail: int = 100,
218        stream: bool = False
219    ) -> str:
220        """
221        Get container logs
222
223        Args:
224            container_id: Container ID or name
225            tail: Number of lines to retrieve
226            stream: Stream logs continuously
227
228        Returns:
229            Container logs as string
230        """
231        try:
232            container = self.client.containers.get(container_id)
233
234            logs = container.logs(
235                tail=tail,
236                stream=stream,
237                timestamps=True
238            )
239
240            if stream:
241                return logs
242            else:
243                return logs.decode('utf-8')
244
245        except docker.errors.NotFound:
246            logger.error(f"Container not found: {container_id}")
247            return ""
248        except docker.errors.APIError as e:
249            logger.error(f"Failed to get logs: {e}")
250            return ""
251
252    def execute_in_container(
253        self,
254        container_id: str,
255        command: str,
256        workdir: Optional[str] = None,
257        user: Optional[str] = None
258    ) -> Dict[str, Any]:
259        """
260        Execute command inside container
261
262        Args:
263            container_id: Container ID or name
264            command: Command to execute
265            workdir: Working directory
266            user: User to execute as
267
268        Returns:
269            Execution result with exit code and output
270        """
271        try:
272            container = self.client.containers.get(container_id)
273
274            logger.info(f"Executing in {container.name}: {command}")
275
276            exec_result = container.exec_run(
277                cmd=command,
278                workdir=workdir,
279                user=user,
280                demux=True
281            )
282
283            stdout, stderr = exec_result.output
284
285            result = {
286                'exit_code': exec_result.exit_code,
287                'stdout': stdout.decode('utf-8') if stdout else '',
288                'stderr': stderr.decode('utf-8') if stderr else ''
289            }
290
291            logger.info(f"  Exit code: {result['exit_code']}")
292
293            return result
294
295        except docker.errors.NotFound:
296            logger.error(f"Container not found: {container_id}")
297            return {'exit_code': -1, 'stdout': '', 'stderr': 'Container not found'}
298        except docker.errors.APIError as e:
299            logger.error(f"Execution failed: {e}")
300            return {'exit_code': -1, 'stdout': '', 'stderr': str(e)}
301
302    def get_container_stats(
303        self,
304        container_id: str
305    ) -> Dict[str, Any]:
306        """
307        Get container resource usage statistics
308
309        Args:
310            container_id: Container ID or name
311
312        Returns:
313            Resource statistics
314        """
315        try:
316            container = self.client.containers.get(container_id)
317
318            stats = container.stats(stream=False)
319
320            # Calculate CPU percentage
321            cpu_delta = stats['cpu_stats']['cpu_usage']['total_usage'] - \
322                        stats['precpu_stats']['cpu_usage']['total_usage']
323            system_delta = stats['cpu_stats']['system_cpu_usage'] - \
324                          stats['precpu_stats']['system_cpu_usage']
325            cpu_percent = (cpu_delta / system_delta) * 100.0 if system_delta > 0 else 0.0
326
327            # Calculate memory usage
328            mem_usage = stats['memory_stats'].get('usage', 0)
329            mem_limit = stats['memory_stats'].get('limit', 1)
330            mem_percent = (mem_usage / mem_limit) * 100.0
331
332            return {
333                'cpu_percent': round(cpu_percent, 2),
334                'memory_usage_mb': round(mem_usage / (1024 * 1024), 2),
335                'memory_limit_mb': round(mem_limit / (1024 * 1024), 2),
336                'memory_percent': round(mem_percent, 2),
337                'network_rx_bytes': stats['networks']['eth0']['rx_bytes'],
338                'network_tx_bytes': stats['networks']['eth0']['tx_bytes']
339            }
340
341        except docker.errors.NotFound:
342            logger.error(f"Container not found: {container_id}")
343            return {}
344        except Exception as e:
345            logger.error(f"Failed to get stats: {e}")
346            return {}
347
348
349# Usage example
350manager = DockerContainerManager()
351
352# Run nginx container
353nginx = manager.run_container(
354    image='nginx:latest',
355    name='web-server',
356    ports={'80/tcp': 8080},
357    environment={
358        'NGINX_HOST': 'example.com',
359        'NGINX_PORT': '80'
360    },
361    volumes={
362        '/host/path': {'bind': '/container/path', 'mode': 'rw'}
363    },
364    cpu_limit=1.0,
365    mem_limit='512m'
366)
367
368# List all containers
369containers = manager.list_containers(all=True)
370
371# Execute command
372result = manager.execute_in_container(
373    container_id='web-server',
374    command='nginx -v'
375)
376print(f"Nginx version: {result['stdout']}")
377
378# Get container stats
379stats = manager.get_container_stats('web-server')
380print(f"CPU usage: {stats['cpu_percent']}%")
381print(f"Memory usage: {stats['memory_usage_mb']} MB ({stats['memory_percent']}%)")
382
383# Get logs
384logs = manager.get_container_logs('web-server', tail=50)
385print(f"Recent logs:\n{logs}")
386
387# Stop and remove
388manager.stop_container('web-server')
389manager.remove_container('web-server')

1import docker
2import tarfile
3import io
4from pathlib import Path
5from typing import Dict, List, Optional, Generator
6import json
7
8
9class DockerImageBuilder:
10    """
11    Build and manage Docker images programmatically
12    """
13
14    def __init__(self):
15        """
16        Initialize Docker client
17        """
18        self.client = docker.from_env()
19        logger.info("Docker Image Builder initialized")
20
21    def build_image(
22        self,
23        path: str,
24        tag: str,
25        dockerfile: str = 'Dockerfile',
26        buildargs: Optional[Dict[str, str]] = None,
27        nocache: bool = False,
28        rm: bool = True,
29        pull: bool = False
30    ) -> Image:
31        """
32        Build Docker image from Dockerfile
33
34        Args:
35            path: Build context path
36            tag: Image tag (e.g., 'myapp:latest')
37            dockerfile: Dockerfile name
38            buildargs: Build arguments
39            nocache: Don't use cache
40            rm: Remove intermediate containers
41            pull: Always pull base image
42
43        Returns:
44            Built Image instance
45        """
46        logger.info(f"Building image: {tag}")
47        logger.info(f"  Context: {path}")
48        logger.info(f"  Dockerfile: {dockerfile}")
49
50        try:
51            image, build_logs = self.client.images.build(
52                path=path,
53                tag=tag,
54                dockerfile=dockerfile,
55                buildargs=buildargs or {},
56                nocache=nocache,
57                rm=rm,
58                pull=pull,
59                quiet=False
60            )
61
62            # Process build logs
63            for log in build_logs:
64                if 'stream' in log:
65                    logger.info(f"  {log['stream'].strip()}")
66                elif 'error' in log:
67                    logger.error(f"  Build error: {log['error']}")
68
69            logger.info(f"✓ Image built successfully: {tag}")
70            logger.info(f"  Image ID: {image.short_id}")
71            logger.info(f"  Size: {image.attrs['Size'] / (1024*1024):.2f} MB")
72
73            return image
74
75        except docker.errors.BuildError as e:
76            logger.error(f"Build failed: {e}")
77            raise
78        except docker.errors.APIError as e:
79            logger.error(f"Docker API error: {e}")
80            raise
81
82    def build_from_string(
83        self,
84        dockerfile_content: str,
85        tag: str,
86        buildargs: Optional[Dict[str, str]] = None
87    ) -> Image:
88        """
89        Build image from Dockerfile string
90
91        Args:
92            dockerfile_content: Dockerfile as string
93            tag: Image tag
94            buildargs: Build arguments
95
96        Returns:
97            Built Image instance
98        """
99        logger.info(f"Building image from string: {tag}")
100
101        # Create in-memory tar archive
102        dockerfile_obj = io.BytesIO(dockerfile_content.encode('utf-8'))
103
104        f = io.BytesIO()
105        tar = tarfile.open(mode='w', fileobj=f)
106
107        dfinfo = tarfile.TarInfo('Dockerfile')
108        dfinfo.size = len(dockerfile_obj.getvalue())
109        tar.addfile(dfinfo, dockerfile_obj)
110
111        tar.close()
112        f.seek(0)
113
114        try:
115            image, build_logs = self.client.images.build(
116                fileobj=f,
117                tag=tag,
118                buildargs=buildargs or {},
119                custom_context=True
120            )
121
122            logger.info(f"✓ Image built: {tag}")
123
124            return image
125
126        except docker.errors.BuildError as e:
127            logger.error(f"Build failed: {e}")
128            raise
129
130    def list_images(
131        self,
132        name: Optional[str] = None,
133        filters: Optional[Dict] = None
134    ) -> List[Image]:
135        """
136        List Docker images
137
138        Args:
139            name: Filter by image name
140            filters: Additional filters
141
142        Returns:
143            List of Image instances
144        """
145        images = self.client.images.list(name=name, filters=filters)
146
147        logger.info(f"Found {len(images)} images")
148
149        for image in images:
150            tags = image.tags[0] if image.tags else '<none>'
151            size_mb = image.attrs['Size'] / (1024 * 1024)
152            logger.info(f"  {image.short_id}: {tags} ({size_mb:.2f} MB)")
153
154        return images
155
156    def tag_image(
157        self,
158        image_id: str,
159        repository: str,
160        tag: str = 'latest'
161    ) -> bool:
162        """
163        Tag an existing image
164
165        Args:
166            image_id: Source image ID or name
167            repository: Target repository
168            tag: Target tag
169
170        Returns:
171            True if tagged successfully
172        """
173        try:
174            image = self.client.images.get(image_id)
175
176            success = image.tag(repository, tag=tag)
177
178            if success:
179                logger.info(f"✓ Image tagged: {repository}:{tag}")
180            else:
181                logger.error(f"Failed to tag image")
182
183            return success
184
185        except docker.errors.ImageNotFound:
186            logger.error(f"Image not found: {image_id}")
187            return False
188        except docker.errors.APIError as e:
189            logger.error(f"Tagging failed: {e}")
190            return False
191
192    def push_image(
193        self,
194        repository: str,
195        tag: str = 'latest',
196        auth_config: Optional[Dict] = None
197    ) -> bool:
198        """
199        Push image to registry
200
201        Args:
202            repository: Repository name
203            tag: Image tag
204            auth_config: Registry authentication
205
206        Returns:
207            True if pushed successfully
208        """
209        full_name = f"{repository}:{tag}"
210
211        logger.info(f"Pushing image: {full_name}")
212
213        try:
214            push_logs = self.client.images.push(
215                repository=repository,
216                tag=tag,
217                auth_config=auth_config,
218                stream=True,
219                decode=True
220            )
221
222            for log in push_logs:
223                if 'status' in log:
224                    logger.info(f"  {log['status']}")
225                if 'error' in log:
226                    logger.error(f"  Push error: {log['error']}")
227                    return False
228
229            logger.info(f"✓ Image pushed: {full_name}")
230
231            return True
232
233        except docker.errors.APIError as e:
234            logger.error(f"Push failed: {e}")
235            return False
236
237    def pull_image(
238        self,
239        repository: str,
240        tag: str = 'latest',
241        auth_config: Optional[Dict] = None
242    ) -> Image:
243        """
244        Pull image from registry
245
246        Args:
247            repository: Repository name
248            tag: Image tag
249            auth_config: Registry authentication
250
251        Returns:
252            Pulled Image instance
253        """
254        full_name = f"{repository}:{tag}"
255
256        logger.info(f"Pulling image: {full_name}")
257
258        try:
259            image = self.client.images.pull(
260                repository=repository,
261                tag=tag,
262                auth_config=auth_config
263            )
264
265            logger.info(f"✓ Image pulled: {full_name}")
266            logger.info(f"  Image ID: {image.short_id}")
267
268            return image
269
270        except docker.errors.ImageNotFound:
271            logger.error(f"Image not found: {full_name}")
272            raise
273        except docker.errors.APIError as e:
274            logger.error(f"Pull failed: {e}")
275            raise
276
277    def remove_image(
278        self,
279        image_id: str,
280        force: bool = False,
281        noprune: bool = False
282    ) -> bool:
283        """
284        Remove Docker image
285
286        Args:
287            image_id: Image ID or name
288            force: Force removal
289            noprune: Don't delete untagged parents
290
291        Returns:
292            True if removed successfully
293        """
294        try:
295            self.client.images.remove(
296                image=image_id,
297                force=force,
298                noprune=noprune
299            )
300
301            logger.info(f"✓ Image removed: {image_id}")
302
303            return True
304
305        except docker.errors.ImageNotFound:
306            logger.error(f"Image not found: {image_id}")
307            return False
308        except docker.errors.APIError as e:
309            logger.error(f"Removal failed: {e}")
310            return False
311
312    def prune_images(
313        self,
314        filters: Optional[Dict] = None
315    ) -> Dict[str, Any]:
316        """
317        Remove unused images
318
319        Args:
320            filters: Filter images to prune
321
322        Returns:
323            Pruning results
324        """
325        logger.info("Pruning unused images...")
326
327        result = self.client.images.prune(filters=filters)
328
329        deleted_count = len(result.get('ImagesDeleted', []))
330        space_reclaimed = result.get('SpaceReclaimed', 0) / (1024 * 1024)
331
332        logger.info(f"✓ Pruned {deleted_count} images")
333        logger.info(f"  Space reclaimed: {space_reclaimed:.2f} MB")
334
335        return result
336
337    def inspect_image(
338        self,
339        image_id: str
340    ) -> Dict[str, Any]:
341        """
342        Get detailed image information
343
344        Args:
345            image_id: Image ID or name
346
347        Returns:
348            Image attributes
349        """
350        try:
351            image = self.client.images.get(image_id)
352
353            info = {
354                'id': image.short_id,
355                'tags': image.tags,
356                'size_mb': image.attrs['Size'] / (1024 * 1024),
357                'created': image.attrs['Created'],
358                'architecture': image.attrs['Architecture'],
359                'os': image.attrs['Os'],
360                'labels': image.attrs['Config'].get('Labels', {}),
361                'env': image.attrs['Config'].get('Env', []),
362                'cmd': image.attrs['Config'].get('Cmd', []),
363                'exposed_ports': list(image.attrs['Config'].get('ExposedPorts', {}).keys())
364            }
365
366            return info
367
368        except docker.errors.ImageNotFound:
369            logger.error(f"Image not found: {image_id}")
370            return {}
371
372
373# Usage example
374builder = DockerImageBuilder()
375
376# Build from Dockerfile
377dockerfile_content = """
378FROM python:3.9-slim
379
380WORKDIR /app
381
382COPY requirements.txt .
383RUN pip install --no-cache-dir -r requirements.txt
384
385COPY . .
386
387EXPOSE 8000
388
389CMD ["python", "app.py"]
390"""
391
392image = builder.build_from_string(
393    dockerfile_content=dockerfile_content,
394    tag='myapp:latest',
395    buildargs={'APP_VERSION': '1.0.0'}
396)
397
398# List images
399images = builder.list_images(name='myapp')
400
401# Tag for registry
402builder.tag_image(
403    image_id='myapp:latest',
404    repository='registry.example.com/myapp',
405    tag='1.0.0'
406)
407
408# Push to registry (with authentication)
409auth_config = {
410    'username': 'user',
411    'password': 'pass'
412}
413
414builder.push_image(
415    repository='registry.example.com/myapp',
416    tag='1.0.0',
417    auth_config=auth_config
418)
419
420# Inspect image
421info = builder.inspect_image('myapp:latest')
422print(f"Image size: {info['size_mb']:.2f} MB")
423print(f"Exposed ports: {info['exposed_ports']}")
424
425# Cleanup
426builder.prune_images()

1import docker
2import yaml
3from pathlib import Path
4from typing import Dict, List, Optional, Any
5import subprocess
6
7
8class DockerComposeManager:
9    """
10    Manage Docker Compose applications
11    """
12
13    def __init__(self, compose_file: str = 'docker-compose.yml'):
14        """
15        Initialize Compose manager
16
17        Args:
18            compose_file: Path to compose file
19        """
20        self.compose_file = Path(compose_file)
21        self.client = docker.from_env()
22
23        logger.info(f"Docker Compose Manager initialized: {compose_file}")
24
25    def generate_compose_file(
26        self,
27        services: Dict[str, Dict],
28        networks: Optional[Dict] = None,
29        volumes: Optional[Dict] = None,
30        version: str = '3.8'
31    ) -> Path:
32        """
33        Generate Docker Compose file
34
35        Args:
36            services: Service definitions
37            networks: Network definitions
38            volumes: Volume definitions
39            version: Compose file version
40
41        Returns:
42            Path to generated file
43        """
44        logger.info("Generating Docker Compose file")
45
46        compose_config = {
47            'version': version,
48            'services': services
49        }
50
51        if networks:
52            compose_config['networks'] = networks
53
54        if volumes:
55            compose_config['volumes'] = volumes
56
57        # Write compose file
58        with open(self.compose_file, 'w') as f:
59            yaml.dump(compose_config, f, default_flow_style=False)
60
61        logger.info(f"✓ Compose file created: {self.compose_file}")
62
63        return self.compose_file
64
65    def up(
66        self,
67        detach: bool = True,
68        build: bool = False,
69        remove_orphans: bool = True
70    ) -> subprocess.CompletedProcess:
71        """
72        Start Compose application
73
74        Args:
75            detach: Run in background
76            build: Build images before starting
77            remove_orphans: Remove containers for services not in compose file
78
79        Returns:
80            Command result
81        """
82        logger.info("Starting Docker Compose application")
83
84        cmd = ['docker-compose', '-f', str(self.compose_file), 'up']
85
86        if detach:
87            cmd.append('-d')
88        if build:
89            cmd.append('--build')
90        if remove_orphans:
91            cmd.append('--remove-orphans')
92
93        result = subprocess.run(
94            cmd,
95            capture_output=True,
96            text=True
97        )
98
99        if result.returncode == 0:
100            logger.info("✓ Application started")
101        else:
102            logger.error(f"Failed to start: {result.stderr}")
103
104        return result
105
106    def down(
107        self,
108        volumes: bool = False,
109        remove_images: Optional[str] = None
110    ) -> subprocess.CompletedProcess:
111        """
112        Stop and remove Compose application
113
114        Args:
115            volumes: Remove volumes
116            remove_images: Remove images ('all' or 'local')
117
118        Returns:
119            Command result
120        """
121        logger.info("Stopping Docker Compose application")
122
123        cmd = ['docker-compose', '-f', str(self.compose_file), 'down']
124
125        if volumes:
126            cmd.append('-v')
127        if remove_images:
128            cmd.extend(['--rmi', remove_images])
129
130        result = subprocess.run(
131            cmd,
132            capture_output=True,
133            text=True
134        )
135
136        if result.returncode == 0:
137            logger.info("✓ Application stopped")
138        else:
139            logger.error(f"Failed to stop: {result.stderr}")
140
141        return result
142
143    def ps(self) -> List[Dict[str, str]]:
144        """
145        List Compose services
146
147        Returns:
148            List of service information
149        """
150        cmd = [
151            'docker-compose',
152            '-f', str(self.compose_file),
153            'ps',
154            '--format', 'json'
155        ]
156
157        result = subprocess.run(
158            cmd,
159            capture_output=True,
160            text=True
161        )
162
163        if result.returncode == 0:
164            services = [
165                json.loads(line)
166                for line in result.stdout.strip().split('\n')
167                if line
168            ]
169
170            logger.info(f"Found {len(services)} services")
171
172            for service in services:
173                logger.info(f"  {service['Name']}: {service['State']}")
174
175            return services
176        else:
177            logger.error(f"Failed to list services: {result.stderr}")
178            return []
179
180    def logs(
181        self,
182        service: Optional[str] = None,
183        follow: bool = False,
184        tail: int = 100
185    ) -> str:
186        """
187        Get Compose service logs
188
189        Args:
190            service: Specific service (None for all)
191            follow: Follow log output
192            tail: Number of lines
193
194        Returns:
195            Log output
196        """
197        cmd = ['docker-compose', '-f', str(self.compose_file), 'logs']
198
199        if follow:
200            cmd.append('-f')
201
202        cmd.extend(['--tail', str(tail)])
203
204        if service:
205            cmd.append(service)
206
207        result = subprocess.run(
208            cmd,
209            capture_output=True,
210            text=True
211        )
212
213        return result.stdout
214
215    def scale(
216        self,
217        service: str,
218        replicas: int
219    ) -> subprocess.CompletedProcess:
220        """
221        Scale a service
222
223        Args:
224            service: Service name
225            replicas: Number of replicas
226
227        Returns:
228            Command result
229        """
230        logger.info(f"Scaling {service} to {replicas} replicas")
231
232        cmd = [
233            'docker-compose',
234            '-f', str(self.compose_file),
235            'up',
236            '-d',
237            '--scale', f'{service}={replicas}'
238        ]
239
240        result = subprocess.run(
241            cmd,
242            capture_output=True,
243            text=True
244        )
245
246        if result.returncode == 0:
247            logger.info(f"✓ Service scaled: {service}={replicas}")
248        else:
249            logger.error(f"Scaling failed: {result.stderr}")
250
251        return result
252
253
254class DockerNetworkManager:
255    """
256    Manage Docker networks
257    """
258
259    def __init__(self):
260        """
261        Initialize network manager
262        """
263        self.client = docker.from_env()
264        logger.info("Docker Network Manager initialized")
265
266    def create_network(
267        self,
268        name: str,
269        driver: str = 'bridge',
270        internal: bool = False,
271        labels: Optional[Dict] = None
272    ) -> docker.models.networks.Network:
273        """
274        Create Docker network
275
276        Args:
277            name: Network name
278            driver: Network driver
279            internal: Make network internal
280            labels: Network labels
281
282        Returns:
283            Network instance
284        """
285        logger.info(f"Creating network: {name}")
286
287        try:
288            network = self.client.networks.create(
289                name=name,
290                driver=driver,
291                internal=internal,
292                labels=labels or {}
293            )
294
295            logger.info(f"✓ Network created: {network.short_id}")
296
297            return network
298
299        except docker.errors.APIError as e:
300            logger.error(f"Network creation failed: {e}")
301            raise
302
303    def list_networks(self) -> List[docker.models.networks.Network]:
304        """
305        List Docker networks
306
307        Returns:
308            List of networks
309        """
310        networks = self.client.networks.list()
311
312        logger.info(f"Found {len(networks)} networks")
313
314        for network in networks:
315            logger.info(f"  {network.short_id}: {network.name} ({network.attrs['Driver']})")
316
317        return networks
318
319    def connect_container(
320        self,
321        network_id: str,
322        container_id: str,
323        aliases: Optional[List[str]] = None
324    ) -> bool:
325        """
326        Connect container to network
327
328        Args:
329            network_id: Network ID or name
330            container_id: Container ID or name
331            aliases: Network aliases
332
333        Returns:
334            True if connected
335        """
336        try:
337            network = self.client.networks.get(network_id)
338
339            network.connect(
340                container=container_id,
341                aliases=aliases or []
342            )
343
344            logger.info(f"✓ Container connected to network")
345
346            return True
347
348        except docker.errors.NotFound:
349            logger.error("Network or container not found")
350            return False
351        except docker.errors.APIError as e:
352            logger.error(f"Connection failed: {e}")
353            return False
354
355
356# Usage example
357compose_manager = DockerComposeManager()
358
359# Define multi-container application
360services = {
361    'web': {
362        'image': 'nginx:latest',
363        'ports': ['80:80'],
364        'depends_on': ['app'],
365        'networks': ['frontend']
366    },
367    'app': {
368        'build': './app',
369        'environment': {
370            'DB_HOST': 'db',
371            'DB_PORT': '5432'
372        },
373        'depends_on': ['db'],
374        'networks': ['frontend', 'backend']
375    },
376    'db': {
377        'image': 'postgres:13',
378        'environment': {
379            'POSTGRES_PASSWORD': 'secret'
380        },
381        'volumes': ['db-data:/var/lib/postgresql/data'],
382        'networks': ['backend']
383    }
384}
385
386networks = {
387    'frontend': {
388        'driver': 'bridge'
389    },
390    'backend': {
391        'driver': 'bridge',
392        'internal': True
393    }
394}
395
396volumes = {
397    'db-data': {}
398}
399
400# Generate compose file
401compose_manager.generate_compose_file(
402    services=services,
403    networks=networks,
404    volumes=volumes
405)
406
407# Start application
408compose_manager.up(build=True)
409
410# List services
411services_list = compose_manager.ps()
412
413# Scale application service
414compose_manager.scale('app', replicas=3)
415
416# Get logs
417logs = compose_manager.logs(service='app', tail=50)
418print(logs)
419
420# Stop application
421compose_manager.down(volumes=False)