circleci_env_cli.py

#!/usr/bin/env python
import logging
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from importlib.metadata import PackageNotFoundError
from importlib.metadata import version as pkg_version

import click
from pycircleci.api import Api, CircleciError

try:
    __version__ = pkg_version("circleci-env-cli")
except PackageNotFoundError:
    __version__ = "unknown"


@dataclass
class _Ctx:
    client: Api
    logger: logging.Logger


def _make_circle_context(api_url: str, token: str, debug: bool = False) -> _Ctx:
    if not token:
        raise SystemExit(
            "Error: CircleCI token is required. Set CIRCLE_TOKEN or use -t/--token."
        )

    if debug:
        logging.basicConfig(
            level=logging.DEBUG,
            format="[%(levelname)s] %(module)s:%(funcName)s - %(message)s",
            stream=sys.stdout,
        )
    else:
        logging.basicConfig(
            level=logging.INFO,
            format="> %(message)s",
            stream=sys.stdout,
        )

    return _Ctx(
        client=Api(token=token, url=api_url),
        logger=logging.getLogger(__name__),
    )


def get_context_id(
    circle_context: _Ctx, org: str, owner_type: str, vcs_type: str, context: str
) -> str | None:
    try:
        contexts = circle_context.client.get_contexts(
            username=org, owner_type=owner_type, vcs_type=vcs_type, paginate=True
        )
        matched = list(filter(lambda c: c["name"] == context, contexts))
    except Exception as error:
        raise SystemExit(f"Error: {error}")

    return matched[0]["id"] if matched else None


def create_context(
    circle_context: _Ctx, org: str, owner_type: str, vcs_type: str, context: str
) -> str:
    answer = input(f'The context named "{context}" was not found. Do you want to create it? [y/n]: ')

    if answer.lower() != "y":
        circle_context.logger.info("Context creation cancelled.")
        sys.exit(0)

    new_context = circle_context.client.add_context(
        context, username=org, owner_type=owner_type, vcs_type=vcs_type
    )
    circle_context.logger.info(f"Successfully created context: {context}")
    return new_context["id"]


def delete_context(circle_context: _Ctx, context_id: str, context: str) -> None:
    confirm = input(f'Are you sure want to delete the context "{context}"? [y/n]: ')

    if confirm.lower() == "y":
        circle_context.client.delete_context(context_id)
        circle_context.logger.info(f"Successfully deleted context: {context}")

    sys.exit(0)


def delete_context_var(circle_context: _Ctx, context_id: str, env: str) -> bool:
    try:
        circle_context.client.delete_context_envvar(context_id, env)
        circle_context.logger.info(f"Successfully deleted context variable: {env}")
        return True
    except Exception as error:
        circle_context.logger.error(f"Error: {error}")
        return False


def upsert_context_var(circle_context: _Ctx, context_id: str, env: str, value: str) -> bool:
    try:
        circle_context.client.add_context_envvar(context_id, env, value)
        circle_context.logger.info(f"Successfully added/updated context variable: {env}")
        return True
    except Exception as error:
        circle_context.logger.error(f"Error: {error}")
        return False


def upsert_project_var(
    circle_context: _Ctx, org: str, vcs_type: str, project: str, env: str, value: str
) -> bool:
    try:
        circle_context.client.add_envvar(org, project, env, value, vcs_type)
        circle_context.logger.info(f"Successfully added/updated variable: {env}")
        return True
    except Exception as error:
        circle_context.logger.error(f"Error: {error}")
        return False


def delete_project_var(
    circle_context: _Ctx, org: str, vcs_type: str, project: str, env: str
) -> bool:
    try:
        circle_context.client.delete_envvar(org, project, env, vcs_type)
        circle_context.logger.info(f"Successfully deleted variable: {env}")
        return True
    except Exception as error:
        circle_context.logger.error(f"Error: {error}")
        return False


def run(
    circle_context: _Ctx,
    owner_type: str,
    context: str,
    project_slug: str,
    env: tuple,
    env_file,
    list_envs: bool,
    delete: bool,
) -> None:
    env_vars = []

    parts = project_slug.split("/")
    if len(parts) == 2:
        vcs_type, org = parts
        project_name = None
    elif len(parts) == 3:
        vcs_type, org, project_name = parts
    else:
        raise SystemExit(
            "Error: Invalid project slug. Expected github/org-name or github/org-name/project-name."
        )

    if not any([env, env_file, list_envs, delete, context]):
        raise SystemExit("Error: nothing to do. Specify -e, -ef, -l, -d, or -c. Use --help for usage.")

    if env:
        env_vars += list(env)

    if env_file:
        seen = set()
        for line in env_file.read().splitlines():
            if line and line not in seen:
                seen.add(line)
                env_vars.append(line)

    context_id = None

    if context:
        context_id = get_context_id(circle_context, org, owner_type, vcs_type, context)

        if not context_id:
            context_id = create_context(circle_context, org, owner_type, vcs_type, context)

        if list_envs:
            context_envs = circle_context.client.get_context_envvars(context_id, True)
            names = sorted(x["variable"] for x in context_envs)
            circle_context.logger.info(f"Context variables ({len(names)}):\n" + "\n".join(names))
            sys.exit(0)

        if not env_vars and delete and context_id:
            delete_context(circle_context, context_id, context)

    if list_envs:
        if not project_name:
            raise SystemExit("Error: project name required for this operation. Use github/org-name/project-name.")
        try:
            envvars = circle_context.client.list_envvars(org, project_name, vcs_type)
            names = sorted(x["name"] for x in envvars)
            circle_context.logger.info(f"Project variables ({len(names)}):\n" + "\n".join(names))
            sys.exit(0)
        except Exception as error:
            circle_context.logger.error(f"Error: {error}")

    with ThreadPoolExecutor(max_workers=5) as executor:
        threadpool = []

        for var in env_vars:
            try:
                key, value = var.split("=", 1)
            except ValueError:
                key, value = (var, "")
            except Exception:
                raise SystemExit(f"Error: failed to proccess variable {var}...")

            key, value = key.strip(), value.strip()

            if context:
                if context_id is None:
                    raise SystemExit(f"Error: could not resolve context '{context}'")
                if delete:
                    threadpool.append(
                        executor.submit(delete_context_var, circle_context, context_id, key)
                    )
                else:
                    threadpool.append(
                        executor.submit(upsert_context_var, circle_context, context_id, key, value)
                    )
            else:
                if not project_name:
                    raise SystemExit(
                        "Error: project name required for this operation. Use github/org-name/project-name."
                    )
                if delete:
                    threadpool.append(
                        executor.submit(delete_project_var, circle_context, org, vcs_type, project_name, key)
                    )
                else:
                    threadpool.append(
                        executor.submit(upsert_project_var, circle_context, org, vcs_type, project_name, key, value)
                    )

        for thread in as_completed(threadpool):
            thread.result()


@click.command(context_settings={"auto_envvar_prefix": "CIRCLE"})
@click.version_option(version=__version__)
@click.option(
    "-u", "--api-url",
    metavar="<circleci_api_url>",
    default="https://circleci.com/api",
    show_default=True,
)
@click.option(
    "-t", "--token",
    metavar="<circleci_token>",
    show_default="CIRCLE_TOKEN",
)
@click.option(
    "-c", "--context",
    metavar="<context_name>",
    help="It will ask for create if does not exists",
)
@click.option(
    "-p", "--project",
    metavar="<project_slug>",
    required=True,
    help="github/org-name for context ops, github/org-name/project-name for project ops",
)
@click.option(
    "-e", "--env",
    multiple=True,
    metavar="<environment_var>",
)
@click.option(
    "-ef", "--env-file",
    metavar="<environment_vars_file>",
    type=click.File("r"),
)
@click.option(
    "-l", "--list-envs",
    metavar="<environment_vars_list>",
    is_flag=True,
)
@click.option(
    "-d", "--delete",
    metavar="<environment_var_delete>",
    is_flag=True,
    help=(
        "Delete mode. Behavior depends on other options:\n\b\n"
        "  -c only:      delete the entire context\n"
        "  -c + -e/-ef:  delete specific context variables\n"
        "  -e/-ef only:  delete specific project variables"
    ),
)
@click.option(
    "-ot", "--owner-type",
    metavar="<owner_type>",
    default="organization",
    show_default=True,
)
@click.option(
    "--debug",
    metavar="<debug>",
    default=False,
    is_flag=True,
)
@click.pass_context
def cli(ctx, api_url, token, context, project, env, env_file, list_envs, delete, owner_type, debug):
    """CLI tool for manage CircleCI contexts and environment vars

    All options can be specified as environment variables in the format:
    CIRCLE_<OPTION>. Example: CIRCLE_TOKEN=********
    """

    try:
        ctx.obj = _make_circle_context(api_url, token, debug)
        run(ctx.obj, owner_type, context, project, env, env_file, list_envs, delete)
    except CircleciError as error:
        raise SystemExit(f"Error: {error}")


if __name__ == "__main__":
    cli()