headquarter/apps/api/src/services/docker/.pi-map.md

apps/api/src/services/docker

dir: apps/api/src/services/docker

index: apps/api/src/services/docker/.pi-map.index.md

role

Provides Docker infrastructure automation for container lifecycle management, service composition, configuration deployment, and secure public tunneling.

files

• __init__.py | Exports Docker service functions for container, compose, config staging, and tunnel operations as a unified package interface. | dep: src.services.docker.compose, src.services.docker.config_staging, src.services.docker.container, src.services.docker.tunnel
• compose.py | Generates, renders, and executes Docker Compose files with volume sorting and template variable substitution. | exp: func:sort_volumes_by_specificity(volumes: list[str]) → list[str], call:vol.split, call:len, call:parts[1].rstrip, call:target.count, call:targets.append, call:Counter(targets).items, call:logger.warning, call:sorted, func:_target_depth(vol: str) → int, call:vol.split, call:len, call:parts[1].rstrip, call:target.count, func:render_compose_template(template: str, variables: dict[str, Any]) → str, call:variables.items, call:result.replace, call:str, call:aliases.items, call:variables.get, func:write_compose_file(instance_dir: str, content: str) → str, call:Path, call:compose_path.write_text, call:str, func:execute_compose_command(compose_path: str, action: str, timeout, env_file) → tuple[int, str, str], call:Path, call:cmd.extend, call:cmd.append, call:subprocess.run, call:str, raise:ValueError | dep: logging, subprocess, collections, pathlib, typing, collections.Counter, pathlib.Path, typing.Any
• config_staging.py | Stages configuration files into instance directories with path security validation and environment file generation. | exp: func:ensure_instance_directory(instance_id: str, base_path) → str, call:Settings, call:Path, call:instance_dir.mkdir, call:str, call:instance_dir.absolute, func:write_env_file(instance_dir: str, env_vars: dict[str, str]) → str, call:Path, call:env_vars.items, call:env_path.write_text, call:"\n".join, call:str, func:write_config_files(instance_dir: str, files: dict[str, str]) → None, call:Path, call:files.items, call:full_path.resolve().relative_to, call:instance_path.resolve, call:full_path.parent.mkdir, call:full_path.write_text, raise:ValueError | dep: logging, pathlib, src.config, src.config.Settings
• container.py | Provides utility functions for querying Docker container runtime state, managing network connections, and finding free ports via subprocess calls to the Docker CLI. | exp: func:get_container_id(instance_name: str) → str | None, call:instance_name.lower, call:subprocess.run, call:result.stdout.strip, call:ps_result.stdout.strip().splitlines, call:line.split, call:len, call:name.lower, func:get_container_name(instance_name: str) → str | None, call:subprocess.run, call:instance_name.lower, call:result.stdout.strip().lstrip, func:get_backend_network_name() → str, call:subprocess.run, call:result.stdout.strip().split, call:net.lower, func:connect_container_to_network(container_name: str, network_name) → bool, call:get_backend_network_name, call:subprocess.run, func:get_container_ip_on_network(container_id: str, network_name) → str | None, call:get_backend_network_name, call:subprocess.run, call:result.stdout.strip, func:is_container_on_network(container_id: str, network_name) → bool, call:get_backend_network_name, call:subprocess.run, func:get_container_status(container_id: str) → dict[str, Any], call:subprocess.run, call:result.stdout.strip().split, call:int, call:len, call:parts[1].isdigit, func:wait_for_container_running(container_id: str, timeout, interval) → dict[str, Any], call:time.time, call:get_container_status, call:time.sleep, func:get_container_logs(container_id: str, tail) → str, call:subprocess.run, call:str, func:find_free_port(start, end) → int, call:range, call:socket.socket, call:s.connect_ex, raise:RuntimeError | dep: logging, subprocess, time, typing, socket
• tunnel.py | Manages Cloudflare tunnels by orchestrating cloudflared Docker containers to expose internal services via temporary public URLs. | exp: func:_tunnel_container_name(instance_name: str) → str, call:instance_name.lower, func:_ensure_image() → None, call:subprocess.run, call:result.stdout.strip, call:logger.info, call:logger.warning, func:_cleanup_stale_tunnel(tunnel_name: str) → None, call:subprocess.run, func:_get_tunnel_logs(tunnel_name: str) → tuple[str, str], call:subprocess.run, func:_get_tunnel_exit_code(tunnel_name: str) → int | None, call:subprocess.run, call:int, call:result.stdout.strip, func:start_tunnel(instance_name: str, container_port: int, timeout, target_url) → dict[str, str], call:_ensure_image, call:_tunnel_container_name, call:_cleanup_stale_tunnel, call:instance_name.lower, call:get_backend_network_name, call:logger.debug, call:" ".join, call:subprocess.run, call:proc.stdout.strip, call:re.compile, call:import("time").time, call:_get_tunnel_logs, call:url_pattern.search, call:match.group, call:_get_tunnel_exit_code, call:import("time").sleep, call:logger.info, raise:RuntimeError, func:stop_tunnel(instance_name: str) → None, call:_tunnel_container_name, call:_cleanup_stale_tunnel, call:logger.debug, func:recreate_tunnel(instance_name: str, container_port: int, target_url) → dict[str, str], call:stop_tunnel, call:start_tunnel, func:check_tunnel_health(url: str, timeout) → dict[str, Any], call:subprocess.run, call:int, call:result.stdout.strip, call:str(exc).lower, call:any | dep: logging, re, subprocess, typing, src.services.docker.container, time

arch

Modular service-oriented architecture with CLI-wrapper pattern for Docker operations, template-based file generation, and subprocess-based external tool orchestration.

tags

tunnel, container, call:subprocess.run, get, name, call:, network, call:result.stdout.strip

symbols

• sort_volumes_by_specificity
• _target_depth
• render_compose_template
• write_compose_file
• execute_compose_command
• ensure_instance_directory
• write_env_file
• write_config_files

workflows

• change docker behavior read: __init__.py, compose.py, config_staging.py
• change docker config read: config_staging.py

dirty