Edit on GitHub

sqlmesh.utils.config 

 1from typing import Any, Optional, Set
 2
 3from sqlmesh.core.config.connection import ConnectionConfig
 4from sqlmesh.utils import yaml
 5
 6
 7# Fields that should be excluded from the configuration hash
 8excluded_fields: Set[str] = {
 9    "concurrent_tasks",
10    "pre_ping",
11    "register_comments",
12}
13
14# Sensitive fields that should be masked in the configuration print or hash
15sensitive_fields: Set[str] = {
16    "access_token",
17    "api_key",
18    "auth_token",
19    "client_secret",
20    "certificate",
21    "credentials",
22    "user",
23    "password",
24    "keytab",
25    "keyfile",
26    "keyfile_json",
27    "principal",
28    "private_key",
29    "private_key_passphrase",
30    "private_key_path",
31    "refresh_token",
32    "secret",
33    "ssh_key",
34    "token",
35}
36
37
38def is_sensitive_field(field_name: str, sensitive_fields: Set[str]) -> bool:
39    """
40    Check if a field name contains any sensitive keywords
41    """
42    field_lower = field_name.lower()
43    return any(sensitive in field_lower for sensitive in sensitive_fields)
44
45
46def mask_sensitive_value(value: Any) -> str:
47    """
48    Mask sensitive values with a placeholder
49    Returns '****' for non-empty values and '' for empty ones
50    """
51    if value and str(value).strip():
52        return "****"
53    return "None"
54
55
56def print_config(config: Optional[ConnectionConfig], console: Any, title: str) -> None:
57    """
58    Print configuration while masking sensitive information
59
60    Args:
61        config: Pydantic model containing configuration
62        console: Console object with log_status_update method
63    """
64    if not config:
65        return
66
67    config_dict = config.dict(mode="json")
68
69    for field_name in config_dict:
70        if is_sensitive_field(field_name, sensitive_fields):
71            config_dict[field_name] = mask_sensitive_value(config_dict[field_name])
72
73    configWithTitle = {title: config_dict}
74    yaml_output = yaml.dump(configWithTitle)
75
76    console.log_status_update(yaml_output)
excluded_fields: Set[str] = {'pre_ping', 'register_comments', 'concurrent_tasks'}
sensitive_fields: Set[str] = {'private_key_path', 'auth_token', 'keyfile', 'api_key', 'secret', 'password', 'client_secret', 'keytab', 'token', 'principal', 'private_key_passphrase', 'certificate', 'credentials', 'access_token', 'ssh_key', 'user', 'keyfile_json', 'refresh_token', 'private_key'}
def is_sensitive_field(field_name: str, sensitive_fields: Set[str]) -> bool: 
39def is_sensitive_field(field_name: str, sensitive_fields: Set[str]) -> bool:
40    """
41    Check if a field name contains any sensitive keywords
42    """
43    field_lower = field_name.lower()
44    return any(sensitive in field_lower for sensitive in sensitive_fields)

Check if a field name contains any sensitive keywords

def mask_sensitive_value(value: Any) -> str: 
47def mask_sensitive_value(value: Any) -> str:
48    """
49    Mask sensitive values with a placeholder
50    Returns '****' for non-empty values and '' for empty ones
51    """
52    if value and str(value).strip():
53        return "****"
54    return "None"

Mask sensitive values with a placeholder Returns '**' for non-empty values and '' for empty ones