|
| 1 | +import json |
| 2 | +import os |
| 3 | +import re |
| 4 | +from pathlib import Path |
| 5 | + |
1 | 6 | import click |
| 7 | +from dotenv import load_dotenv |
2 | 8 |
|
3 | 9 | from datapilot.core.mcp_utils.mcp import mcp |
4 | 10 | from datapilot.core.platforms.dbt.cli.cli import dbt |
5 | 11 |
|
6 | 12 |
|
| 13 | +def load_config_from_file(): |
| 14 | + """Load configuration from ~/.altimate/altimate.json if it exists.""" |
| 15 | + config_path = Path.home() / ".altimate" / "altimate.json" |
| 16 | + |
| 17 | + if not config_path.exists(): |
| 18 | + return {} |
| 19 | + |
| 20 | + try: |
| 21 | + with open(config_path, 'r') as f: |
| 22 | + config = json.load(f) |
| 23 | + return config |
| 24 | + except (json.JSONDecodeError, IOError) as e: |
| 25 | + click.echo(f"Warning: Failed to load config from {config_path}: {e}", err=True) |
| 26 | + return {} |
| 27 | + |
| 28 | + |
| 29 | +def substitute_env_vars(value): |
| 30 | + """Replace ${env:ENV_VARIABLE} patterns with actual environment variable values.""" |
| 31 | + if not isinstance(value, str): |
| 32 | + return value |
| 33 | + |
| 34 | + # Pattern to match ${env:VARIABLE_NAME} |
| 35 | + pattern = r'\$\{env:([^}]+)\}' |
| 36 | + |
| 37 | + def replacer(match): |
| 38 | + env_var = match.group(1) |
| 39 | + return os.environ.get(env_var, match.group(0)) |
| 40 | + |
| 41 | + return re.sub(pattern, replacer, value) |
| 42 | + |
| 43 | + |
| 44 | +def process_config(config): |
| 45 | + """Process configuration dictionary to substitute environment variables.""" |
| 46 | + processed = {} |
| 47 | + for key, value in config.items(): |
| 48 | + processed[key] = substitute_env_vars(value) |
| 49 | + return processed |
| 50 | + |
| 51 | + |
7 | 52 | @click.group() |
8 | 53 | @click.option("--token", required=False, help="Your API token for authentication.") |
9 | 54 | @click.option("--instance-name", required=False, help="Your tenant ID.") |
10 | 55 | @click.option("--backend-url", required=False, help="Altimate's Backend URL", default="https://api.myaltimate.com") |
11 | 56 | @click.pass_context |
12 | 57 | def datapilot(ctx, token, instance_name, backend_url): |
13 | 58 | """Altimate CLI for DBT project management.""" |
14 | | - # Store common options in context |
| 59 | + # Load .env file from current directory if it exists |
| 60 | + load_dotenv() |
| 61 | + |
| 62 | + # Load configuration from file |
| 63 | + file_config = load_config_from_file() |
| 64 | + file_config = process_config(file_config) |
| 65 | + |
| 66 | + # Map config file keys to CLI option names |
| 67 | + config_mapping = { |
| 68 | + 'altimateApiKey': 'token', |
| 69 | + 'altimateInstanceName': 'instance_name', |
| 70 | + 'altimateUrl': 'backend_url' |
| 71 | + } |
| 72 | + |
| 73 | + # Store common options in context, with CLI args taking precedence |
15 | 74 | ctx.ensure_object(dict) |
16 | | - ctx.obj['token'] = token |
17 | | - ctx.obj['instance_name'] = instance_name |
18 | | - ctx.obj['backend_url'] = backend_url |
| 75 | + |
| 76 | + # Apply file config first |
| 77 | + for file_key, cli_key in config_mapping.items(): |
| 78 | + if file_key in file_config: |
| 79 | + ctx.obj[cli_key] = file_config[file_key] |
| 80 | + |
| 81 | + # Override with CLI arguments if provided |
| 82 | + if token is not None: |
| 83 | + ctx.obj['token'] = token |
| 84 | + if instance_name is not None: |
| 85 | + ctx.obj['instance_name'] = instance_name |
| 86 | + if backend_url != "https://api.myaltimate.com": # Only override if not default |
| 87 | + ctx.obj['backend_url'] = backend_url |
| 88 | + |
| 89 | + # Set defaults if nothing was provided |
| 90 | + ctx.obj.setdefault('token', None) |
| 91 | + ctx.obj.setdefault('instance_name', None) |
| 92 | + ctx.obj.setdefault('backend_url', 'https://api.myaltimate.com') |
19 | 93 |
|
20 | 94 |
|
21 | 95 | datapilot.add_command(dbt) |
|
0 commit comments