1023 lines
35 KiB
Python
1023 lines
35 KiB
Python
import asyncio
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from importlib import import_module
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import click
|
||
|
||
from cli.metadata import (
|
||
COMPLETION_SHELL_SPEC_REGISTRY,
|
||
ENVIRONMENT_OPTION_SERVICE,
|
||
CompletionShellSpecRegistry,
|
||
EnvironmentOptionService,
|
||
)
|
||
|
||
DEFAULT_ALEMBIC_REVISION_CHOICES = ('head', 'base', 'current', '-1')
|
||
DYNAMIC_COMPLETION_TIMEOUT_SECONDS = 0.8
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CompletionContextResolver:
|
||
"""
|
||
shell completion 上下文解析器。
|
||
|
||
该对象负责解析当前补全场景下的环境、缓存名称和项目目录，
|
||
将原本散落的参数与环境变量解析逻辑收口到单一职责对象中。
|
||
"""
|
||
|
||
def resolve_completion_env(self, ctx: click.Context | None) -> str:
|
||
"""
|
||
解析当前 completion 场景使用的环境名称。
|
||
|
||
:param ctx: Click 上下文
|
||
:return: 环境名称，默认返回 `dev`
|
||
"""
|
||
if ctx is not None:
|
||
env_value = getattr(ctx, 'params', {}).get('env')
|
||
if isinstance(env_value, str) and env_value.strip():
|
||
return env_value.strip()
|
||
|
||
comp_words = os.environ.get('COMP_WORDS', '').strip()
|
||
if comp_words:
|
||
tokens = comp_words.split()
|
||
for index, token in enumerate(tokens):
|
||
if token.startswith('--env='):
|
||
env_value = token.split('=', 1)[1].strip()
|
||
if env_value:
|
||
return env_value
|
||
if token == '--env' and index + 1 < len(tokens):
|
||
env_value = tokens[index + 1].strip()
|
||
if env_value:
|
||
return env_value
|
||
|
||
return 'dev'
|
||
|
||
def resolve_cache_name_for_completion(self, ctx: click.Context | None) -> str:
|
||
"""
|
||
解析当前 cache key 补全场景使用的缓存名称。
|
||
|
||
:param ctx: Click 上下文
|
||
:return: 缓存名称，缺失时返回空字符串
|
||
"""
|
||
if ctx is not None:
|
||
cache_name = getattr(ctx, 'params', {}).get('cache_name')
|
||
if isinstance(cache_name, str) and cache_name.strip():
|
||
return cache_name.strip()
|
||
|
||
comp_words = os.environ.get('COMP_WORDS', '').strip()
|
||
if not comp_words:
|
||
return ''
|
||
|
||
tokens = comp_words.split()
|
||
try:
|
||
cache_index = tokens.index('cache')
|
||
except ValueError:
|
||
return ''
|
||
|
||
if cache_index + 2 >= len(tokens):
|
||
return ''
|
||
|
||
subcommand = tokens[cache_index + 1]
|
||
if subcommand not in {'get', 'ttl'}:
|
||
return ''
|
||
|
||
cache_name = tokens[cache_index + 2].strip()
|
||
if cache_name.startswith('-'):
|
||
return ''
|
||
return cache_name
|
||
|
||
@staticmethod
|
||
def resolve_project_dir() -> Path:
|
||
"""
|
||
获取当前 CLI 工作目录对应的项目根目录。
|
||
|
||
:return: 当前项目根目录
|
||
"""
|
||
return Path.cwd().resolve()
|
||
|
||
@staticmethod
|
||
def normalize_completion_prefix(incomplete: str) -> str:
|
||
"""
|
||
规范化补全输入前缀。
|
||
|
||
:param incomplete: 原始未完成输入
|
||
:return: 规范化后的前缀文本
|
||
"""
|
||
return incomplete.strip()
|
||
|
||
@staticmethod
|
||
def to_display_path(path: Path, *, project_dir: Path) -> str:
|
||
"""
|
||
将绝对路径转换为适合补全显示的路径文本。
|
||
|
||
:param path: 原始路径
|
||
:param project_dir: 项目根目录
|
||
:return: 用于 shell completion 的显示路径
|
||
"""
|
||
try:
|
||
relative_path = path.relative_to(project_dir)
|
||
return str(relative_path) or '.'
|
||
except ValueError:
|
||
return str(path)
|
||
|
||
|
||
@dataclass
|
||
class DynamicCompletionService:
|
||
"""
|
||
动态补全执行服务。
|
||
|
||
该服务负责 runtime 模块导入、只读协程执行、标准 payload 提取与
|
||
候选过滤，从而避免各补全函数继续复制相同的模板。
|
||
|
||
:param context_resolver: completion 上下文解析器
|
||
:param timeout_seconds: 动态补全超时时间
|
||
"""
|
||
|
||
context_resolver: CompletionContextResolver = field(default_factory=CompletionContextResolver)
|
||
timeout_seconds: float = DYNAMIC_COMPLETION_TIMEOUT_SECONDS
|
||
|
||
@staticmethod
|
||
def _scoped_app_env(env: str) -> Any:
|
||
"""
|
||
为动态补全临时注入 `APP_ENV` 环境变量。
|
||
|
||
`config.env` 在导入阶段会优先读取 `APP_ENV`，因此补全场景下
|
||
直接切换进程环境变量比改写 `sys.argv` 更稳定，也更符合当前
|
||
后端配置模块的解析约束。
|
||
|
||
:param env: 当前补全环境名称
|
||
:return: 环境变量恢复上下文
|
||
"""
|
||
|
||
class _CompletionEnvScope:
|
||
def __enter__(self) -> None:
|
||
self._original_app_env = os.environ.get('APP_ENV')
|
||
os.environ['APP_ENV'] = env
|
||
|
||
def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
|
||
if self._original_app_env is None:
|
||
os.environ.pop('APP_ENV', None)
|
||
else:
|
||
os.environ['APP_ENV'] = self._original_app_env
|
||
|
||
return _CompletionEnvScope()
|
||
|
||
def load_runtime_module(self, module_name: str) -> Any:
|
||
"""
|
||
加载动态补全依赖的 runtime 模块。
|
||
|
||
:param module_name: runtime 模块名
|
||
:return: 已导入模块
|
||
"""
|
||
return import_module(module_name)
|
||
|
||
def run_completion_coroutine(self, coroutine: Any, *, env: str) -> Any:
|
||
"""
|
||
在补全场景下执行异步只读查询，并临时注入 `APP_ENV`。
|
||
|
||
:param coroutine: 待执行协程
|
||
:param env: 当前补全环境名称
|
||
:return: 协程执行结果
|
||
"""
|
||
try:
|
||
with self._scoped_app_env(env):
|
||
return asyncio.run(asyncio.wait_for(coroutine, timeout=self.timeout_seconds))
|
||
finally:
|
||
if asyncio.iscoroutine(coroutine):
|
||
coroutine.close()
|
||
|
||
@staticmethod
|
||
def extract_completion_items(payload: dict[str, Any], field_name: str) -> list[str]:
|
||
"""
|
||
从 CLI 标准结果中提取补全候选字段列表。
|
||
|
||
:param payload: CLI 标准结果字典
|
||
:param field_name: 需要提取的字段名
|
||
:return: 去重后的字符串列表
|
||
"""
|
||
if not isinstance(payload, dict) or not payload.get('ok', False):
|
||
return []
|
||
|
||
items = payload.get('items')
|
||
if not isinstance(items, list):
|
||
page_payload = payload.get('page')
|
||
if isinstance(page_payload, dict):
|
||
items = page_payload.get('rows')
|
||
if not isinstance(items, list):
|
||
return []
|
||
|
||
candidates = []
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
value = item.get(field_name)
|
||
if isinstance(value, str) and value.strip():
|
||
candidates.append(value.strip())
|
||
return sorted(set(candidates))
|
||
|
||
@staticmethod
|
||
def extract_completion_values(payload: dict[str, Any], field_name: str) -> list[str]:
|
||
"""
|
||
从 CLI 标准结果中提取字符串或整数类型的补全候选值。
|
||
|
||
:param payload: CLI 标准结果字典
|
||
:param field_name: 需要提取的字段名
|
||
:return: 去重后的字符串列表
|
||
"""
|
||
if not isinstance(payload, dict) or not payload.get('ok', False):
|
||
return []
|
||
|
||
items = payload.get('items')
|
||
if not isinstance(items, list):
|
||
page_payload = payload.get('page')
|
||
if isinstance(page_payload, dict):
|
||
items = page_payload.get('rows')
|
||
if not isinstance(items, list):
|
||
return []
|
||
|
||
candidates = []
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
value = item.get(field_name)
|
||
if isinstance(value, str) and value.strip():
|
||
candidates.append(value.strip())
|
||
elif isinstance(value, int):
|
||
candidates.append(str(value))
|
||
return sorted(set(candidates))
|
||
|
||
@staticmethod
|
||
def extract_completion_list(payload: dict[str, Any], field_name: str) -> list[str]:
|
||
"""
|
||
从 CLI 标准结果中提取直接位于顶层字段的字符串列表。
|
||
|
||
该方法用于缓存键名等不走 `items/page.rows` 结构的场景，
|
||
使这类动态补全也能统一复用 runtime 对象加载与协程执行逻辑。
|
||
|
||
:param payload: CLI 标准结果字典
|
||
:param field_name: 需要提取的顶层字段名
|
||
:return: 去重后的字符串列表
|
||
"""
|
||
if not isinstance(payload, dict) or not payload.get('ok', False):
|
||
return []
|
||
|
||
items = payload.get(field_name)
|
||
if not isinstance(items, list):
|
||
return []
|
||
|
||
candidates = [item.strip() for item in items if isinstance(item, str) and item.strip()]
|
||
return sorted(set(candidates))
|
||
|
||
@staticmethod
|
||
def filter_candidates(candidates: list[str], incomplete: str) -> list[str]:
|
||
"""
|
||
按未完成输入前缀过滤候选值。
|
||
|
||
:param candidates: 原始候选列表
|
||
:param incomplete: 当前未完成输入
|
||
:return: 过滤后的候选列表
|
||
"""
|
||
normalized_incomplete = incomplete.strip().lower()
|
||
return [candidate for candidate in candidates if candidate.lower().startswith(normalized_incomplete)]
|
||
|
||
def complete_dynamic_items(
|
||
self,
|
||
ctx: click.Context | None,
|
||
incomplete: str,
|
||
*,
|
||
runtime_module_name: str,
|
||
runtime_object_name: str | None = None,
|
||
runtime_method_name: str,
|
||
runtime_kwargs: dict[str, Any],
|
||
field_name: str,
|
||
allow_int_values: bool = False,
|
||
) -> list[str]:
|
||
"""
|
||
执行通用的动态只读补全查询。
|
||
|
||
:param ctx: Click 上下文
|
||
:param incomplete: 当前未完成输入
|
||
:param runtime_module_name: runtime 模块名
|
||
:param runtime_object_name: runtime 对象名，为空时直接使用模块
|
||
:param runtime_method_name: runtime 方法名
|
||
:param runtime_kwargs: runtime 调用参数
|
||
:param field_name: 待提取字段名
|
||
:param allow_int_values: 是否允许整数候选
|
||
:return: 过滤后的候选列表
|
||
"""
|
||
env = self.context_resolver.resolve_completion_env(ctx)
|
||
try:
|
||
runtime_module = self.load_runtime_module(runtime_module_name)
|
||
runtime_target = getattr(runtime_module, runtime_object_name) if runtime_object_name else runtime_module
|
||
runtime_method = getattr(runtime_target, runtime_method_name)
|
||
payload = self.run_completion_coroutine(runtime_method(**runtime_kwargs), env=env)
|
||
except Exception:
|
||
return []
|
||
|
||
if allow_int_values:
|
||
candidates = self.extract_completion_values(payload, field_name)
|
||
else:
|
||
candidates = self.extract_completion_items(payload, field_name)
|
||
return self.filter_candidates(candidates, incomplete)
|
||
|
||
def complete_dynamic_list(
|
||
self,
|
||
ctx: click.Context | None,
|
||
incomplete: str,
|
||
*,
|
||
runtime_module_name: str,
|
||
runtime_object_name: str | None = None,
|
||
runtime_method_name: str,
|
||
runtime_args: tuple[Any, ...] = (),
|
||
runtime_kwargs: dict[str, Any] | None = None,
|
||
field_name: str,
|
||
) -> list[str]:
|
||
"""
|
||
执行返回顶层字符串列表字段的动态只读补全查询。
|
||
|
||
:param ctx: Click 上下文
|
||
:param incomplete: 当前未完成输入
|
||
:param runtime_module_name: runtime 模块名
|
||
:param runtime_object_name: runtime 对象名，为空时直接使用模块
|
||
:param runtime_method_name: runtime 方法名
|
||
:param runtime_args: runtime 位置参数
|
||
:param runtime_kwargs: runtime 关键字参数
|
||
:param field_name: 顶层列表字段名
|
||
:return: 过滤后的候选列表
|
||
"""
|
||
env = self.context_resolver.resolve_completion_env(ctx)
|
||
try:
|
||
runtime_module = self.load_runtime_module(runtime_module_name)
|
||
runtime_target = getattr(runtime_module, runtime_object_name) if runtime_object_name else runtime_module
|
||
runtime_method = getattr(runtime_target, runtime_method_name)
|
||
payload = self.run_completion_coroutine(
|
||
runtime_method(*runtime_args, **(runtime_kwargs or {})),
|
||
env=env,
|
||
)
|
||
except Exception:
|
||
return []
|
||
|
||
candidates = self.extract_completion_list(payload, field_name)
|
||
return self.filter_candidates(candidates, incomplete)
|
||
|
||
|
||
@dataclass
|
||
class StaticCompletionProvider:
|
||
"""
|
||
静态 completion 提供器。
|
||
|
||
该对象负责 shell、环境、缓存名称以及 Alembic revision 等静态或
|
||
本地推导型补全候选。
|
||
|
||
:param context_resolver: completion 上下文解析器
|
||
:param dynamic_service: 动态补全执行服务
|
||
:param shell_spec_registry: shell 元数据注册表
|
||
:param environment_option_service: 环境选项服务
|
||
"""
|
||
|
||
context_resolver: CompletionContextResolver = field(default_factory=CompletionContextResolver)
|
||
dynamic_service: DynamicCompletionService = field(
|
||
default_factory=lambda: DynamicCompletionService(context_resolver=CompletionContextResolver())
|
||
)
|
||
shell_spec_registry: CompletionShellSpecRegistry = field(default_factory=lambda: COMPLETION_SHELL_SPEC_REGISTRY)
|
||
environment_option_service: EnvironmentOptionService = field(default_factory=lambda: ENVIRONMENT_OPTION_SERVICE)
|
||
|
||
def list_completion_shells(self) -> list[str]:
|
||
"""
|
||
获取已声明的 completion shell 列表。
|
||
|
||
:return: shell 名称列表
|
||
"""
|
||
return self.shell_spec_registry.list_shell_names()
|
||
|
||
def list_static_cache_names(self) -> list[str]:
|
||
"""
|
||
读取系统内置缓存名称列表。
|
||
|
||
:return: 缓存名称列表
|
||
"""
|
||
try:
|
||
redis_init_key_config = self.dynamic_service.load_runtime_module('common.enums').RedisInitKeyConfig
|
||
except Exception:
|
||
return []
|
||
return [key_config.key for key_config in redis_init_key_config if getattr(key_config, 'key', '')]
|
||
|
||
def complete_env_values(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为 `--env` 选项提供可补全的环境名称。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 匹配的环境名称列表
|
||
"""
|
||
del ctx, args
|
||
normalized_incomplete = incomplete.strip().lower()
|
||
return [
|
||
env_name
|
||
for env_name in self.environment_option_service.discover_env_names()
|
||
if env_name.lower().startswith(normalized_incomplete)
|
||
]
|
||
|
||
def complete_shell_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为 shell 参数提供静态补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: shell 名称列表
|
||
"""
|
||
del ctx, args
|
||
normalized_incomplete = incomplete.strip().lower()
|
||
return [
|
||
shell_name for shell_name in self.list_completion_shells() if shell_name.startswith(normalized_incomplete)
|
||
]
|
||
|
||
def complete_cache_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为缓存名称参数提供静态补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 缓存名称列表
|
||
"""
|
||
del ctx, args
|
||
normalized_incomplete = incomplete.strip().lower()
|
||
return [
|
||
cache_name
|
||
for cache_name in sorted(self.list_static_cache_names())
|
||
if cache_name.lower().startswith(normalized_incomplete)
|
||
]
|
||
|
||
def complete_alembic_revisions(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为数据库迁移版本参数提供本地补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 迁移版本候选列表
|
||
"""
|
||
del ctx, args
|
||
normalized_incomplete = self.context_resolver.normalize_completion_prefix(incomplete).lower()
|
||
revision_choices = set(DEFAULT_ALEMBIC_REVISION_CHOICES)
|
||
versions_dir = self.context_resolver.resolve_project_dir() / 'alembic' / 'versions'
|
||
if versions_dir.is_dir():
|
||
for revision_file in versions_dir.glob('*.py'):
|
||
revision_stem = revision_file.stem
|
||
revision_id = revision_stem.split('-', 1)[-1].split('_', 1)[0].strip()
|
||
if revision_id:
|
||
revision_choices.add(revision_id)
|
||
return sorted(choice for choice in revision_choices if choice.lower().startswith(normalized_incomplete))
|
||
|
||
|
||
@dataclass
|
||
class PathCompletionProvider:
|
||
"""
|
||
路径 completion 提供器。
|
||
|
||
该对象负责项目相对 SQL 文件路径和输出路径补全。
|
||
|
||
:param context_resolver: completion 上下文解析器
|
||
"""
|
||
|
||
context_resolver: CompletionContextResolver = field(default_factory=CompletionContextResolver)
|
||
|
||
def complete_sql_files(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为 SQL 文件参数提供项目内 `.sql` 文件补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: SQL 文件路径列表
|
||
"""
|
||
del ctx, args
|
||
normalized_incomplete = self.context_resolver.normalize_completion_prefix(incomplete).lower()
|
||
project_dir = self.context_resolver.resolve_project_dir()
|
||
candidates = []
|
||
for sql_file in project_dir.rglob('*.sql'):
|
||
if '.git' in sql_file.parts or '__pycache__' in sql_file.parts:
|
||
continue
|
||
display_path = self.context_resolver.to_display_path(sql_file, project_dir=project_dir)
|
||
if display_path.lower().startswith(normalized_incomplete):
|
||
candidates.append(display_path)
|
||
return sorted(candidates)
|
||
|
||
def complete_output_paths(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为导出文件参数提供目录和 `.zip` 文件补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 目录或 zip 文件路径列表
|
||
"""
|
||
del ctx, args
|
||
raw_incomplete = self.context_resolver.normalize_completion_prefix(incomplete)
|
||
project_dir = self.context_resolver.resolve_project_dir()
|
||
|
||
expanded_input = Path(raw_incomplete).expanduser()
|
||
input_is_absolute = expanded_input.is_absolute()
|
||
if raw_incomplete.endswith(('/', os.sep)):
|
||
search_dir = expanded_input if input_is_absolute else (project_dir / expanded_input).resolve()
|
||
partial_name = ''
|
||
else:
|
||
search_dir = (
|
||
expanded_input.parent
|
||
if input_is_absolute
|
||
else (
|
||
project_dir / expanded_input.parent if str(expanded_input.parent) != '.' else project_dir
|
||
).resolve()
|
||
)
|
||
partial_name = expanded_input.name
|
||
|
||
if not search_dir.exists() or not search_dir.is_dir():
|
||
return []
|
||
|
||
candidates = []
|
||
for child_path in sorted(search_dir.iterdir()):
|
||
if not child_path.name.startswith(partial_name):
|
||
continue
|
||
if child_path.is_dir():
|
||
display_path = self.context_resolver.to_display_path(child_path, project_dir=project_dir)
|
||
candidates.append(f'{display_path}/')
|
||
continue
|
||
if child_path.suffix.lower() == '.zip':
|
||
candidates.append(self.context_resolver.to_display_path(child_path, project_dir=project_dir))
|
||
return candidates
|
||
|
||
|
||
@dataclass
|
||
class DomainDynamicCompletionProvider:
|
||
"""
|
||
业务域动态 completion 提供器。
|
||
|
||
该对象负责配置、代码生成、缓存、任务等运行时驱动的只读补全。
|
||
|
||
:param context_resolver: completion 上下文解析器
|
||
:param dynamic_service: 动态补全执行服务
|
||
"""
|
||
|
||
context_resolver: CompletionContextResolver = field(default_factory=CompletionContextResolver)
|
||
dynamic_service: DynamicCompletionService = field(init=False)
|
||
|
||
def __post_init__(self) -> None:
|
||
"""
|
||
初始化动态补全执行服务。
|
||
|
||
:return: None
|
||
"""
|
||
self.dynamic_service = DynamicCompletionService(context_resolver=self.context_resolver)
|
||
|
||
def complete_config_keys(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为参数键名提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 参数键名列表
|
||
"""
|
||
del args
|
||
return self.dynamic_service.complete_dynamic_items(
|
||
ctx,
|
||
incomplete,
|
||
runtime_module_name='cli.runtime.config',
|
||
runtime_object_name='CONFIG_RUNTIME',
|
||
runtime_method_name='list_configs',
|
||
runtime_kwargs={'config_key': incomplete, 'paged': True, 'page_num': 1, 'page_size': 20},
|
||
field_name='configKey',
|
||
)
|
||
|
||
def complete_gen_table_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为代码生成业务表名称提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 业务表名称列表
|
||
"""
|
||
del args
|
||
return self.dynamic_service.complete_dynamic_items(
|
||
ctx,
|
||
incomplete,
|
||
runtime_module_name='cli.runtime.gen',
|
||
runtime_object_name='GEN_RUNTIME',
|
||
runtime_method_name='list_gen_tables',
|
||
runtime_kwargs={'table_name': incomplete, 'paged': True, 'page_num': 1, 'page_size': 20},
|
||
field_name='tableName',
|
||
)
|
||
|
||
def complete_gen_db_table_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为数据库物理表名称提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 数据库物理表名称列表
|
||
"""
|
||
del args
|
||
return self.dynamic_service.complete_dynamic_items(
|
||
ctx,
|
||
incomplete,
|
||
runtime_module_name='cli.runtime.gen',
|
||
runtime_object_name='GEN_RUNTIME',
|
||
runtime_method_name='list_gen_db_tables',
|
||
runtime_kwargs={'table_name': incomplete, 'paged': True, 'page_num': 1, 'page_size': 20},
|
||
field_name='tableName',
|
||
)
|
||
|
||
def complete_cache_keys(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为缓存键名提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 缓存键名列表
|
||
"""
|
||
del args
|
||
cache_name = self.context_resolver.resolve_cache_name_for_completion(ctx)
|
||
if not cache_name:
|
||
return []
|
||
return self.dynamic_service.complete_dynamic_list(
|
||
ctx,
|
||
incomplete,
|
||
runtime_module_name='cli.runtime.cache',
|
||
runtime_object_name='CACHE_RUNTIME',
|
||
runtime_method_name='list_cache_keys',
|
||
runtime_args=(cache_name,),
|
||
field_name='keys',
|
||
)
|
||
|
||
def complete_job_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为任务名称提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 任务名称列表
|
||
"""
|
||
del args
|
||
return self.dynamic_service.complete_dynamic_items(
|
||
ctx,
|
||
incomplete,
|
||
runtime_module_name='cli.runtime.job',
|
||
runtime_object_name='JOB_RUNTIME',
|
||
runtime_method_name='list_jobs',
|
||
runtime_kwargs={'job_name': incomplete, 'paged': True, 'page_num': 1, 'page_size': 20},
|
||
field_name='jobName',
|
||
)
|
||
|
||
def complete_job_ids(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为任务 ID 提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 任务 ID 列表
|
||
"""
|
||
del args
|
||
return self.dynamic_service.complete_dynamic_items(
|
||
ctx,
|
||
incomplete,
|
||
runtime_module_name='cli.runtime.job',
|
||
runtime_object_name='JOB_RUNTIME',
|
||
runtime_method_name='list_jobs',
|
||
runtime_kwargs={'paged': True, 'page_num': 1, 'page_size': 20},
|
||
field_name='jobId',
|
||
allow_int_values=True,
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class CompletionProviderRegistry:
|
||
"""
|
||
completion 提供器注册表。
|
||
|
||
该注册表只负责持有静态补全、路径补全与动态只读补全 provider，
|
||
供上层 gateway、安装器与其他协作者按职责选择使用。
|
||
|
||
:param context_resolver: completion 上下文解析器
|
||
:param dynamic_service: 动态补全执行服务
|
||
:param static_provider: 静态 completion 提供器
|
||
:param path_provider: 路径 completion 提供器
|
||
:param domain_provider: 业务域动态 completion 提供器
|
||
"""
|
||
|
||
context_resolver: CompletionContextResolver = field(default_factory=CompletionContextResolver)
|
||
dynamic_service: DynamicCompletionService = field(init=False)
|
||
static_provider: StaticCompletionProvider = field(init=False)
|
||
path_provider: PathCompletionProvider = field(init=False)
|
||
domain_provider: DomainDynamicCompletionProvider = field(init=False)
|
||
|
||
def __post_init__(self) -> None:
|
||
"""
|
||
初始化聚合 provider 依赖。
|
||
|
||
:return: None
|
||
"""
|
||
self.dynamic_service = DynamicCompletionService(context_resolver=self.context_resolver)
|
||
self.static_provider = StaticCompletionProvider(
|
||
context_resolver=self.context_resolver,
|
||
dynamic_service=self.dynamic_service,
|
||
)
|
||
self.path_provider = PathCompletionProvider(context_resolver=self.context_resolver)
|
||
self.domain_provider = DomainDynamicCompletionProvider(context_resolver=self.context_resolver)
|
||
self.domain_provider.dynamic_service = self.dynamic_service
|
||
|
||
|
||
@dataclass
|
||
class CompletionProviderGateway:
|
||
"""
|
||
completion 对外网关。
|
||
|
||
该对象负责对命令声明层、上下文层和 TUI 层暴露统一的 completion 入口，
|
||
避免 `CompletionProviderRegistry` 本体继续膨胀为大而全的委托门面。
|
||
|
||
:param provider_registry: completion provider 注册表
|
||
"""
|
||
|
||
provider_registry: CompletionProviderRegistry = field(default_factory=CompletionProviderRegistry)
|
||
|
||
def list_completion_shells(self) -> list[str]:
|
||
"""
|
||
获取已声明的 completion shell 列表。
|
||
|
||
:return: shell 名称列表
|
||
"""
|
||
return self.provider_registry.static_provider.list_completion_shells()
|
||
|
||
def list_static_cache_names(self) -> list[str]:
|
||
"""
|
||
读取系统内置缓存名称列表。
|
||
|
||
:return: 缓存名称列表
|
||
"""
|
||
return self.provider_registry.static_provider.list_static_cache_names()
|
||
|
||
def complete_env_values(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为 `--env` 选项提供可补全的环境名称。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 匹配的环境名称列表
|
||
"""
|
||
return self.provider_registry.static_provider.complete_env_values(ctx, args, incomplete)
|
||
|
||
def complete_shell_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为 shell 参数提供静态补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: shell 名称列表
|
||
"""
|
||
return self.provider_registry.static_provider.complete_shell_names(ctx, args, incomplete)
|
||
|
||
def complete_cache_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为缓存名称参数提供静态补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 缓存名称列表
|
||
"""
|
||
return self.provider_registry.static_provider.complete_cache_names(ctx, args, incomplete)
|
||
|
||
def complete_alembic_revisions(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为数据库迁移版本参数提供本地补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 迁移版本候选列表
|
||
"""
|
||
return self.provider_registry.static_provider.complete_alembic_revisions(ctx, args, incomplete)
|
||
|
||
def complete_sql_files(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为 SQL 文件参数提供项目内 `.sql` 文件补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: SQL 文件路径列表
|
||
"""
|
||
return self.provider_registry.path_provider.complete_sql_files(ctx, args, incomplete)
|
||
|
||
def complete_output_paths(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为导出文件参数提供目录和 `.zip` 文件补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 目录或 zip 文件路径列表
|
||
"""
|
||
return self.provider_registry.path_provider.complete_output_paths(ctx, args, incomplete)
|
||
|
||
def complete_config_keys(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为参数键名提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 参数键名列表
|
||
"""
|
||
return self.provider_registry.domain_provider.complete_config_keys(ctx, args, incomplete)
|
||
|
||
def complete_gen_table_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为代码生成业务表名称提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 业务表名称列表
|
||
"""
|
||
return self.provider_registry.domain_provider.complete_gen_table_names(ctx, args, incomplete)
|
||
|
||
def complete_gen_db_table_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为数据库物理表名称提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 数据库物理表名称列表
|
||
"""
|
||
return self.provider_registry.domain_provider.complete_gen_db_table_names(ctx, args, incomplete)
|
||
|
||
def complete_cache_keys(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为缓存键名提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 缓存键名列表
|
||
"""
|
||
return self.provider_registry.domain_provider.complete_cache_keys(ctx, args, incomplete)
|
||
|
||
def complete_job_names(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为任务名称提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 任务名称列表
|
||
"""
|
||
return self.provider_registry.domain_provider.complete_job_names(ctx, args, incomplete)
|
||
|
||
def complete_job_ids(
|
||
self,
|
||
ctx: click.Context | None,
|
||
args: list[str] | None,
|
||
incomplete: str,
|
||
) -> list[str]:
|
||
"""
|
||
为任务 ID 提供动态只读补全结果。
|
||
|
||
:param ctx: Click 上下文
|
||
:param args: 当前命令参数列表
|
||
:param incomplete: 当前未完成输入片段
|
||
:return: 任务 ID 列表
|
||
"""
|
||
return self.provider_registry.domain_provider.complete_job_ids(ctx, args, incomplete)
|
||
|
||
|
||
COMPLETION_PROVIDER_REGISTRY = CompletionProviderRegistry()
|
||
COMPLETION_PROVIDER_GATEWAY = CompletionProviderGateway(provider_registry=COMPLETION_PROVIDER_REGISTRY)
|