add config argument and env var

src/mu/cli/aws.py

@@ -3,7 +3,7 @@
 
 import click
 
-from ..config import Config, cli_load
+from ..config import Config
 from ..libs import api_gateway, auth, ec2, ecr, ecs, gateway
 from .core import cli
 
@@ -21,9 +21,10 @@ def aws():
 @click.option('--name-prefix', help='Filter on name tag')
 @click.option('--name-key', help='Key of tag to use for name', default='Name')
 @click.option('--verbose', '-v', is_flag=True)
-def subnets(target_env, name_prefix, name_key, verbose):
+@click.pass_context
+def subnets(ctx: click.Context, target_env, name_prefix, name_key, verbose):
     """List ec2 subnets"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
     b3_sess = auth.b3_sess(config.aws_region)
 
     for name, subnet in ec2.describe_subnets(b3_sess, name_prefix, name_key).items():
@@ -36,9 +37,10 @@ def subnets(target_env, name_prefix, name_key, verbose):
 @click.argument('only_names', nargs=-1)
 @click.option('--env', 'target_env')
 @click.option('--verbose', '-v', is_flag=True)
-def security_groups(target_env, only_names, verbose):
+@click.pass_context
+def security_groups(ctx: click.Context, target_env, only_names, verbose):
     """List ec2 subnets"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
     b3_sess = auth.b3_sess(config.aws_region)
 
     for name, group in ec2.describe_security_groups(b3_sess, only_names).items():
@@ -50,9 +52,10 @@ def security_groups(target_env, only_names, verbose):
 @aws.command()
 @click.option('--env', 'target_env')
 @click.option('--verbose', '-v', is_flag=True)
-def ecs_clusters(target_env, verbose):
+@click.pass_context
+def ecs_clusters(ctx: click.Context, target_env, verbose):
     """List App Runner instance configurations"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
     b3_sess = auth.b3_sess(config.aws_region)
 
     ecs_ = ecs.ECS(b3_sess)
@@ -65,7 +68,7 @@ def ecs_clusters(target_env, verbose):
 @click.pass_context
 def ecr_push(ctx: click.Context, target_env: str | None):
     """Push built image to ecr"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
     repo_name = config.resource_ident
     print(config.aws_region)
     repos = ecr.Repos(auth.b3_sess(config.aws_region))
@@ -77,9 +80,10 @@ def ecr_push(ctx: click.Context, target_env: str | None):
 @cli.command()
 @click.argument('target_env', required=False)
 @click.option('--verbose', is_flag=True)
-def ecr_repos(verbose: bool, target_env: str | None):
+@click.pass_context
+def ecr_repos(ctx: click.Context, verbose: bool, target_env: str | None):
     """List ECR repos in active account"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
     b3_sess = auth.b3_sess(config.aws_region)
 
     repos = ecr.Repos(b3_sess)
@@ -94,9 +98,10 @@ def ecr_repos(verbose: bool, target_env: str | None):
 @click.argument('repo_name', required=False)
 @click.option('--verbose', is_flag=True)
 @click.option('--env', 'target_env')
-def ecr_images(verbose: bool, target_env: str | None, repo_name: str | None):
+@click.pass_context
+def ecr_images(ctx: click.Context, verbose: bool, target_env: str | None, repo_name: str | None):
     """List all images in a repo"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
     b3_sess = auth.b3_sess(config.aws_region)
 
     repos = ecr.Repos(b3_sess)
@@ -130,7 +135,7 @@ def ecr_tags(
     repo_name: str | None,
 ):
     """List ecr tags"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
     b3_sess = auth.b3_sess(config.aws_region)
 
     repos = ecr.Repos(b3_sess)
@@ -145,9 +150,10 @@ def ecr_tags(
 
 @aws.command()
 @click.option('--verbose', is_flag=True)
-def api_gateways(verbose: bool):
+@click.pass_context
+def api_gateways(ctx: click.Context, verbose: bool):
     """List api gateways in active account"""
-    config: Config = cli_load(None)
+    config: Config = ctx.obj['load_config'](None)
     b3_sess = auth.b3_sess(config.aws_region)
 
     apis = api_gateway.APIs(b3_sess)

src/mu/cli/core.py

@@ -1,9 +1,10 @@
+from pathlib import Path
 from pprint import pprint
 
 import click
 
 import mu.config
-from mu.config import Config, cli_load
+from mu.config import Config, default_env, load
 from mu.libs import auth, logs, sqs, sts, utils
 from mu.libs.lamb import Lambda
 from mu.libs.status import Status
@@ -13,16 +14,31 @@
 
 
 @click.group()
+@click.option(
+    '--config',
+    'config_path',
+    type=click.Path(path_type=Path),
+    help='Path to mu config file',
+    envvar='MU_CONFIG_PATH',
+)
 @logs.click_options
-def cli(log_level: str):
+@click.pass_context
+def cli(ctx: click.Context, log_level: str, config_path: Path | None):
     logs.init_logging(log_level)
+    ctx.ensure_object(dict)
+
+    def load_config(env: str | None = None) -> Config:
+        return load(Path.cwd(), env or default_env(), config_path)
+
+    ctx.obj['load_config'] = load_config
 
 
 @cli.command()
 @click.argument('target_env', required=False)
-def auth_check(target_env):
+@click.pass_context
+def auth_check(ctx: click.Context, target_env):
     """Check AWS auth by displaying account info"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
     b3_sess = auth.b3_sess(config.aws_region)
     ident: str = sts.caller_identity(b3_sess)
     print('Account:', ident['Account'])
@@ -42,9 +58,10 @@ def auth_check(target_env):
 @cli.command()
 @click.argument('target_env', required=False)
 @click.option('--resolve-env', is_flag=True, help='Show env after resolution (e.g. secrets)')
-def config(target_env: str, resolve_env: bool):
+@click.pass_context
+def config(ctx: click.Context, target_env: str, resolve_env: bool):
     """Display mu config for active project"""
-    config: Config = cli_load(target_env)
+    config: Config = ctx.obj['load_config'](target_env)
 
     sess = auth.b3_sess(config.aws_region)
     config.apply_sess(sess)
@@ -54,24 +71,25 @@ def config(target_env: str, resolve_env: bool):
 
 @cli.command()
 @click.argument('envs', nargs=-1)
-def provision(envs: list[str]):
+@click.pass_context
+def provision(ctx: click.Context, envs: list[str]):
     """Provision lambda function in environment given (or default)"""
     envs = envs or [None]
 
     for env in envs:
-        lamb = Lambda(cli_load(env))
+        lamb = Lambda(ctx.obj['load_config'](env))
         lamb.provision()
 
 
 @cli.command()
 @click.argument('envs', nargs=-1)
 @click.option('--build', is_flag=True)
 @click.pass_context
-def deploy(ctx, envs: list[str], build: bool):
+def deploy(ctx: click.Context, envs: list[str], build: bool):
     """Deploy local image to ecr, update lambda"""
     envs = envs or [mu.config.default_env()]
 
-    configs = [cli_load(env) for env in envs]
+    configs = [ctx.obj['load_config'](env) for env in envs]
 
     if build:
         service_names = [config.compose_service for config in configs]
@@ -85,18 +103,19 @@ def deploy(ctx, envs: list[str], build: bool):
 @cli.command()
 @click.argument('target_env')
 @click.option('--force-repo', is_flag=True)
-def delete(target_env: str, force_repo: bool):
+@click.pass_context
+def delete(ctx: click.Context, target_env: str, force_repo: bool):
     """Delete lambda and optionally related infra"""
-    lamb = Lambda(cli_load(target_env))
+    lamb = Lambda(ctx.obj['load_config'](target_env))
     lamb.delete(target_env, force_repo=force_repo)
 
 
 @cli.command()
 @click.argument('target_env', required=False)
-def build(target_env: str):
+@click.pass_context
+def build(ctx: click.Context, target_env: str):
     """Build lambda container with docker compose"""
-
-    conf = cli_load(target_env)
+    conf = ctx.obj['load_config'](target_env)
     utils.compose_build(conf.compose_service)
 
 
@@ -107,10 +126,16 @@ def build(target_env: str):
 @click.option('--host', default='localhost:8080')
 @click.option('--local', is_flag=True)
 @click.pass_context
-def invoke(ctx, target_env: str, action: str, host: str, action_args: list, local: bool):
+def invoke(
+    ctx: click.Context,
+    target_env: str,
+    action: str,
+    host: str,
+    action_args: list,
+    local: bool,
+):
     """Invoke lambda with diagnostics or given action"""
-
-    lamb = Lambda(cli_load(target_env))
+    lamb = Lambda(ctx.obj['load_config'](target_env))
     if local:
         result = lamb.invoke_rei(host, action, action_args)
     else:
@@ -138,7 +163,7 @@ def _logs(
     if not first and not last:
         last = 10 if streams else 25
 
-    lamb = Lambda(cli_load(target_env))
+    lamb = Lambda(ctx.obj['load_config'](target_env))
     lamb.logs(first, last, streams)
 
 
@@ -164,11 +189,12 @@ def sqs_list(ctx: click.Context, verbose: bool, delete: bool, name_prefix=str):
 @cli.command()
 @click.argument('action', type=click.Choice(('show', 'provision', 'delete')))
 @click.argument('target_env', required=False)
-def domain_name(target_env: str, action: str):
+@click.pass_context
+def domain_name(ctx: click.Context, target_env: str, action: str):
     """Manage AWS config needed for domain name support"""
     from ..libs import gateway
 
-    config: Config = cli_load(target_env or mu.config.default_env())
+    config: Config = ctx.obj['load_config'](target_env)
     assert config.domain_name
 
     gw = gateway.Gateway(config)
@@ -189,8 +215,9 @@ def domain_name(target_env: str, action: str):
 
 @cli.command()
 @click.argument('target_env', required=False)
-def status(target_env: str):
+@click.pass_context
+def status(ctx: click.Context, target_env: str | None):
     """Check status of all infrastructure components for the app"""
-    config = cli_load(target_env or mu.config.default_env())
+    config = ctx.obj['load_config'](target_env)
 
     print(Status.fetch(config))

src/mu/config.py

@@ -173,12 +173,18 @@ def default_env():
     return environ.get('MU_DEFAULT_ENV') or utils.host_user()
 
 
-def load(start_at: Path, env: str) -> Config:
+def load(start_at: Path, env: str, mu_fpath: Path | None = None) -> Config:
     pp_fpath = find_upwards(start_at, 'pyproject.toml')
     if pp_fpath is None:
         raise Exception(f'No pyproject.toml found in {start_at} or parents')
 
-    mu_fpath = pp_fpath.with_name('mu.toml')
+    if mu_fpath:
+        if not mu_fpath.exists():
+            raise Exception(f'Config file not found: {mu_fpath}')
+        mu_fpath = mu_fpath
+    else:
+        mu_fpath = pp_fpath.with_name('mu.toml')
+
     if mu_fpath.exists():
         config_fpath = mu_fpath
         key_prefix = ''
@@ -222,7 +228,3 @@ def load(start_at: Path, env: str) -> Config:
             default=(),
         ),
     )
-
-
-def cli_load(env) -> Config:
-    return load(Path.cwd(), env or default_env())

src/mu_tests/pkg2/mu2.toml

@@ -0,0 +1,3 @@
+project-org = 'Starfleet'
+aws-region = 'us-east-2'
+domain-name = 'pkg2.domain2.com'

src/mu_tests/test_config.py

@@ -38,13 +38,21 @@ def test_minimal_config_defaults(self, m_host_user):
         assert c.function_arn == 'arn:aws:lambda:south:1234:function:starfleet-tng-func-qa'
 
     @mock_patch_obj(config.utils, 'host_user')
-    def test_mu_toml(self, m_host_user):
+    def test_inferred_mu_toml(self, m_host_user):
         m_host_user.return_value = 'picard.science-station'
 
         c = load('pkg2')
         assert c.resource_ident == 'starfleet-tng-lambda-func-qa'
         assert c.domain_name == 'pkg2.example.com'
 
+    @mock_patch_obj(config.utils, 'host_user')
+    def test_specified_mu_toml(self, m_host_user):
+        m_host_user.return_value = 'picard.science-station'
+
+        c = config.load(tests_dpath / 'pkg2', 'qa', tests_dpath / 'pkg2' / 'mu2.toml')
+        assert c.resource_ident == 'starfleet-tng-lambda-func-qa'
+        assert c.domain_name == 'pkg2.domain2.com'
+
     def test_sqs_configs(self):
         conf = load('pkg-sqs')
         sqs = conf.aws_configs('sqs')