Skip to content

Spells (API reference)

This is an API reference for the all the Invoke tasks and helper functions that are included in the Conjuring package.

conjuring.spells.aws

AWS: ECR login.

clean_ecr_url(c, url=None)

Clean an AWS ECR URL.

Source code in src/conjuring/spells/aws.py 
48
49
50
51
52
53
54
def clean_ecr_url(c: Context, url: str | None = None) -> str:
    """Clean an AWS ECR URL."""
    if not url:
        account = fzf_aws_account(c)
        region = fzf_aws_region(c)
        return f"{account}.dkr.ecr.{region}.amazonaws.com"
    return urlparse(url).netloc

ecr_login(c, url='')

Log in to AWS ECR.

Using Amazon ECR with the AWS CLI - Amazon ECR

Source code in src/conjuring/spells/aws.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@task
def ecr_login(c: Context, url: str = "") -> None:
    """Log in to AWS ECR.

    [Using Amazon ECR with the AWS CLI - Amazon ECR](https://docs.aws.amazon.com/AmazonECR/latest/userguide/getting-started-cli.html#cli-authenticate-registry)
    """
    profile = fzf_aws_profile(c)
    url = clean_ecr_url(c, url)
    run_command(
        c,
        "aws ecr get-login-password --profile",
        profile,
        "| docker login --username AWS --password-stdin",
        url,
    )

fzf_aws_account(c)

Select an AWS account from the config file.

Source code in src/conjuring/spells/aws.py 
33
34
35
def fzf_aws_account(c: Context) -> str:
    """Select an AWS account from the config file."""
    return run_with_fzf(c, f"rg -o 'aws:iam::[^:]+' {AWS_CONFIG} | cut -d ':' -f 4 | sort -u")

fzf_aws_profile(c, partial_name=None)

Select an AWS profile from a partial profile name using fzf.

Source code in src/conjuring/spells/aws.py 
24
25
26
27
28
29
30
def fzf_aws_profile(c: Context, partial_name: str | None = None) -> str:
    """Select an AWS profile from a partial profile name using fzf."""
    if not partial_name and (aws_profile := os.environ.get("AWS_PROFILE")) and aws_profile:
        typer.echo(f"Using env variable AWS_PROFILE (set to '{aws_profile}')")
        return aws_profile

    return run_with_fzf(c, LIST_AWS_PROFILES_COMMAND, query=partial_name or "")

fzf_aws_region(c)

Select an AWS region from the config file.

Source code in src/conjuring/spells/aws.py 
38
39
40
def fzf_aws_region(c: Context) -> str:
    """Select an AWS region from the config file."""
    return run_with_fzf(c, f"rg -o '^region.+' {AWS_CONFIG} | tr -d ' ' | cut -d'=' -f 2 | sort -u")

list_aws_profiles(c)

List AWS profiles from the config file.

Source code in src/conjuring/spells/aws.py 
19
20
21
def list_aws_profiles(c: Context) -> list[str]:
    """List AWS profiles from the config file."""
    return run_lines(c, LIST_AWS_PROFILES_COMMAND)

run_aws_vault(c, *pieces, profile=None)

Run AWS vault commands in a subshell, or open a subshell if no commands were provided.

Source code in src/conjuring/spells/aws.py 
43
44
45
def run_aws_vault(c: Context, *pieces: str, profile: str | None = None) -> Result:
    """Run AWS vault commands in a subshell, or open a subshell if no commands were provided."""
    return run_command(c, "aws-vault exec", fzf_aws_profile(c, profile), "--", *pieces, pty=False)

conjuring.spells.direnv

direnv: init local dir.

init(c, source_up=False, dotenv=False, all_=False)

Configure direnv in the local dir.

Source code in src/conjuring/spells/direnv.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@task()
def init(c: Context, source_up: bool = False, dotenv: bool = False, all_: bool = False) -> None:
    """Configure direnv in the local dir."""
    if all_:
        source_up = dotenv = True
    if not (source_up or dotenv or all_):
        print_error("Choose one of the options: --source-up, --dotenv, --all")
        return

    envrc = Path(ENVRC)
    content = envrc.read_text() if envrc.exists() else ""

    if source_up and SOURCE_UP_IF_EXISTS not in content:
        content += SOURCE_UP_IF_EXISTS_TEMPLATE

    if dotenv and DOTENV_IF_EXISTS not in content:
        content += DOTENV_IF_EXISTS_TEMPLATE

    if content:
        envrc.write_text(content)
        c.run("direnv allow")

    bat(c, ".env*")

conjuring.spells.docker

Docker: remove containers and volumes.

rm_containers(c, container='', all_=False, exited=False)

Remove Docker containers.

Source code in src/conjuring/spells/docker.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@task(
    help={
        "container": "Container name to remove (regexp)",
        "all_": "All containers",
        "exited": "Exited containers",
    },
)
def rm_containers(c: Context, container: str = "", all_: bool = False, exited: bool = False) -> None:
    """Remove Docker containers."""
    cmd = []
    if all_:
        cmd = ["docker ps -a"]
    elif exited:
        cmd = ["docker ps -a -f status=exited"]
    elif container:
        cmd = ["docker ps -a | grep -e", container]
    if not cmd:
        print_error("Choose one argument. Run with --help to see available argument")
        return

    run_command(c, *cmd, dry=False)
    run_command(c, *cmd, "| tail +2 | awk '{print $1}' | xargs docker rm -f")
    run_command(c, *cmd)

rm_volumes(c, dangling=False)

Remove Docker volumes.

Source code in src/conjuring/spells/docker.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
@task(help=({"dangling": "Dangling volumes"}))
def rm_volumes(c: Context, dangling: bool = False) -> None:
    """Remove Docker volumes."""
    cmd = ""
    if dangling:
        cmd = 'docker volume ls -f "dangling=true"'
    if not cmd:
        print_error("Choose one argument. Run with --help to see available argument")
        return

    run_command(c, cmd, dry=False)
    run_command(c, f"docker volume rm $({cmd} -q)")
    run_command(c, cmd)

conjuring.spells.duplicity

Backup and restore with Duplicity.

backup(c)

Backup files with Duplicity.

Source code in src/conjuring/spells/duplicity.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@task
def backup(c: Context) -> None:
    """Backup files with Duplicity."""
    host = print_hostname(c)
    backup_dir = f"file://{BACKUP_DIR}/{host}/duplicity/"
    # To back up directly on OneDrive:
    # backup_dir = f"onedrive://Backup/{host}/duplicity/"
    print(f"Backup dir: {backup_dir}")

    print_success("Scanning repos:")
    files_to_append = []
    for line in run_lines(c, f"fd -u -t d --max-depth 2 {DOT_GIT}$ {CODE_DIR} {WOLT_DIR}"):
        repo = Path(line).parent
        if not is_valid_git_repository(repo):
            continue

        print_success(f"Files to keep in repo {repo}:")
        with c.cd(repo):
            for relative_path in run_lines(c, "git keep"):
                abs_path = (repo / relative_path).absolute()
                end_slash = "/" if abs_path.is_dir() else ""

                keep_file_path = f"{abs_path}{end_slash}"
                files_to_append.append(keep_file_path)

                print_func = print_warning if end_slash else print
                print_func(f"  {abs_path.relative_to(Path.home())}")

    template_file = Path("~/dotfiles/backup-duplicity-template.cfg").expanduser()
    print(f"Template file: {template_file}")

    template_contents = template_file.read_text()
    duplicity_config = Template(template_contents).substitute({"HOME": Path.home()})

    with NamedTemporaryFile("r+", delete=False) as temp_file:
        temp_file.write(duplicity_config)
        temp_file.write("\n".join(files_to_append))
        temp_file.flush()
        run_command(
            c,
            "duplicity",
            f"--name='{host}-backup'",
            "-v info",
            f"--include-filelist={temp_file.name}",
            "--exclude='**' $HOME/",
            backup_dir,
        )

list_current_files(c)

List all files included in the current duplicity backup.

Source code in src/conjuring/spells/duplicity.py 
74
75
76
77
78
79
80
81
@task
def list_current_files(c: Context) -> None:
    """List all files included in the current duplicity backup."""
    host = print_hostname(c)
    backup_dir = f"file://{BACKUP_DIR}/{host}/duplicity/"
    print(f"Backup dir: {backup_dir}")

    c.run(f"duplicity list-current-files {backup_dir}")

print_hostname(c)

Print the hostname of the current machine.

Source code in src/conjuring/spells/duplicity.py 
18
19
20
21
22
def print_hostname(c: Context) -> str:
    """Print the hostname of the current machine."""
    host = c.run("hostname | sed 's/.local//'", dry=False).stdout.strip()
    print(f"Host: {host}")
    return host

restore(c)

Restore files with Duplicity. You will be prompted to choose the source dir. Restore dir is ~/Downloads.

Source code in src/conjuring/spells/duplicity.py 
84
85
86
87
88
89
90
91
92
93
@task
def restore(c: Context) -> None:
    """Restore files with Duplicity. You will be prompted to choose the source dir. Restore dir is ~/Downloads."""
    print_hostname(c)
    chosen_dir = run_with_fzf(c, f"fd -d 2 -t d duplicity {BACKUP_DIR}")
    if not chosen_dir:
        return

    source_computer = Path(chosen_dir).parent.name
    c.run(f"duplicity restore file://{chosen_dir} ~/Downloads/duplicity-restore/{source_computer}/")

conjuring.spells.fork

GitHub forks: configure remote and sync.

remote(c, username, remote_='upstream')

Configure a remote for a fork.

Source code in src/conjuring/spells/fork.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@task(
    help={
        "username": "The owner of the original repository. Required",
        "remote": "The remote to sync with (default: upstream)",
    },
)
def remote(c: Context, username: str, remote_: str = "upstream") -> None:
    """[Configure a remote for a fork](https://docs.github.com/en/github/collaborating-with-issues-and-pull-requests/configuring-a-remote-for-a-fork)."""
    if username.startswith("-"):
        msg = "Arguments should be: username [--remote]"
        raise Exit(msg)
    if not remote_:
        remote_ = username

    project = c.run(r"git remote -v | rg origin | head -1 | rg -o '/(.+)\.git' -r '$1'", pty=False).stdout.strip()
    c.run(f"git remote add {remote_} https://github.com/{username}/{project}.git", warn=True)
    c.run("git remote -v")

sync(c, remote_='upstream')

Sync a fork.

Source code in src/conjuring/spells/fork.py 
29
30
31
32
33
34
35
@task(help={"remote": "The remote to sync with (default: upstream)"})
def sync(c: Context, remote_: str = "upstream") -> None:
    """[Sync a fork](https://docs.github.com/en/github/collaborating-with-issues-and-pull-requests/syncing-a-fork)."""
    c.run(f"git fetch {remote_}")
    existing_branch = Git(c).checkout("master", "main")
    c.run(f"git merge {remote_}/{existing_branch}")
    c.run("git push")

conjuring.spells.generic

Generic spells: list to-do items in files.

Location dataclass

Location of a to-do item in a file.

Source code in src/conjuring/spells/generic.py 
46
47
48
49
50
51
52
53
54
55
56
@dataclass
class Location:
    """Location of a to-do item in a file."""

    path: str
    line: int | str
    comment: str

    def __post_init__(self) -> None:
        self.line = int(self.line)
        self.comment = self.comment.strip()

ToDoItem dataclass

A to-do item.

Source code in src/conjuring/spells/generic.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@dataclass(frozen=True)
class ToDoItem:
    """A to-do item."""

    which: str
    assignee: str
    description: str

    @property
    def sort_key(self) -> str:
        """Key to sort the instance.

        String concatenation works.
        Checking both fields separately with ``and`` conditions didn't work: sort order was not as expected
        (meaning fix-me tasks first, then to-do tasks).
        """
        return f"{self.which.casefold()}-{self.assignee.casefold()}-{self.description.casefold()}"

    def __lt__(self, other: ToDoItem) -> bool:
        return self.sort_key < other.sort_key

sort_key property

Key to sort the instance.

String concatenation works. Checking both fields separately with and conditions didn't work: sort order was not as expected (meaning fix-me tasks first, then to-do tasks).

todo(c, cz=False, valid=True, invalid=True, short=False, priority='', markdown=False, dir_='')

List to-dos and fix-mes in code. Optionally check if the description follows Conventional Commits (cz check).

Source code in src/conjuring/spells/generic.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@task(
    help={
        "cz": "Run commitizen (cz check) to validate the description of the to-do item as a commit message",
        "valid": "When using cz check, print valid to-do items",
        "invalid": "When using cz check, print invalid to-do items",
        "short": "Short format: only the description, without the lines of code where to-do items were found",
        "priority": "Specify an assignee and show only higher priority tasks for them"
        f" ({FIX_ME} or {TO_DO}(<assignee>)",
        "markdown": "Print the output in Markdown format",
        "dir": "Partial directory names to search for items. Use multiple times or a comma-separated list",
    },
    iterable=["dir"],
)
def todo(  # noqa: C901,PLR0913,PLR0912
    c: Context,
    cz: bool = False,
    valid: bool = True,
    invalid: bool = True,
    short: bool = False,
    priority: str = "",
    markdown: bool = False,
    dir_: str = "",
) -> None:
    """List to-dos and fix-mes in code. Optionally check if the description follows Conventional Commits (cz check)."""
    dir_names = []
    if dir_:
        for one_dir in always_iterable(dir_):
            if "," in one_dir:
                dir_names.extend(one_dir.split(","))
            else:
                dir_names.append(one_dir)
    all_todos: dict[ToDoItem, list[Location]] = _parse_all_todos(c, priority, dir_names)

    if markdown:
        _print_todos_as_markdown(all_todos, short)
        return

    for item, locations in sorted(all_todos.items()):  # type: ToDoItem, list[Location]
        func = print_success
        if cz:
            result = run_command(c, "cz check -m", quote(item.description), hide=True, warn=True)
            if result.ok:
                if not valid:
                    continue
            else:
                if not invalid:
                    continue
                func = print_error

        assignee_str = f"({item.assignee.upper()})" if item.assignee else ""
        func(f"{item.which}{assignee_str}: {item.description}")

        if short:
            continue
        for loc in locations:  # type: Location
            typer.echo(f"   {loc.path}:{loc.line} {loc.comment}")

conjuring.spells.git

Git: update all, extract subtree, rewrite history, ...

Git

Git helpers.

Source code in src/conjuring/spells/git.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class Git:
    """Git helpers."""

    # Use "tail +2" to remove the blank line at the top
    SHOW_ALL_FILE_HISTORY = 'git log --pretty="format:" --name-only | sort -u | tail +2'

    def __init__(self, context: Context) -> None:
        self.context = context

    def current_branch(self) -> str:
        """Return the current branch name."""
        return run_stdout(self.context, "git branch --show-current")

    def default_branch(self) -> str:
        """Return the default branch name (master/main/develop/development)."""
        return run_stdout(
            self.context,
            "git branch -a | rg -o -e /master -e /develop.+ -e /main | sort -u | cut -b 2- | head -1",
        )

    def checkout(self, *branches: str) -> str:
        """Try checking out the specified branches in order."""
        for branch in branches:
            try:
                self.context.run(f"git checkout {branch}")
            except UnexpectedExit:  # noqa: PERF203
                pass
            else:
                return branch
        return ""

    @property
    def github_username(self) -> str:
        """The GitHub username configured in the global settings."""
        return global_config()["github"]["user"]

    def choose_local_branch(self, branch: str) -> str:
        """Choose a local branch."""
        return run_with_fzf(self.context, "git branch --list | rg -v develop | cut -b 3-", query=branch)

github_username property

The GitHub username configured in the global settings.

checkout(*branches)

Try checking out the specified branches in order.

Source code in src/conjuring/spells/git.py 
73
74
75
76
77
78
79
80
81
82
def checkout(self, *branches: str) -> str:
    """Try checking out the specified branches in order."""
    for branch in branches:
        try:
            self.context.run(f"git checkout {branch}")
        except UnexpectedExit:  # noqa: PERF203
            pass
        else:
            return branch
    return ""

choose_local_branch(branch)

Choose a local branch.

Source code in src/conjuring/spells/git.py 
89
90
91
def choose_local_branch(self, branch: str) -> str:
    """Choose a local branch."""
    return run_with_fzf(self.context, "git branch --list | rg -v develop | cut -b 3-", query=branch)

current_branch()

Return the current branch name.

Source code in src/conjuring/spells/git.py 
62
63
64
def current_branch(self) -> str:
    """Return the current branch name."""
    return run_stdout(self.context, "git branch --show-current")

default_branch()

Return the default branch name (master/main/develop/development).

Source code in src/conjuring/spells/git.py 
66
67
68
69
70
71
def default_branch(self) -> str:
    """Return the default branch name (master/main/develop/development)."""
    return run_stdout(
        self.context,
        "git branch -a | rg -o -e /master -e /develop.+ -e /main | sort -u | cut -b 2- | head -1",
    )

PrefixBranch dataclass

Tuple of prefix and branch name.

Source code in src/conjuring/spells/git.py 
94
95
96
97
98
99
@dataclass(frozen=True)
class PrefixBranch:
    """Tuple of prefix and branch name."""

    prefix: str
    branch: str

body(c, prefix=False, original_order=False)

Prepare a commit body to be used on pull requests and squashed commits.

Source code in src/conjuring/spells/git.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
@task(
    help={
        "prefix": "Keep the Conventional Commits prefix",
        "original_order": "Don't sort bullets, keep them in original order",
    },
)
def body(c: Context, prefix: bool = False, original_order: bool = False) -> None:
    """Prepare a commit body to be used on pull requests and squashed commits."""
    default_branch = set_default_branch(c)
    bullets = []
    for line in run_lines(c, f"git log {default_branch}..", "--format=%s%n%b"):
        clean = line.strip(" -")
        if (
            "Merge branch" in clean
            or "Merge remote-tracking branch" in clean
            or "Revert " in clean
            or "This reverts" in clean
            or not clean
        ):
            continue

        # Remove Jira ticket with regex
        clean = REGEX_JIRA.sub("", clean).replace("()", "").replace("[]", "").strip(" -")

        # Split on the Conventional Commit prefix
        if not prefix and ":" in clean:
            clean = clean.split(":", 1)[1].strip()

        bullets.append(f"- {clean}")

    results = bullets if original_order else sorted(set(bullets))
    typer.echo("\n".join(results))

changes_since_tag(c, tag='', files=False, verbose=False, by_author=False)

Display changes (commits or files) since the last tag (or a chosen tag).

Source code in src/conjuring/spells/git.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
@task(
    help={
        "tag": "Name of the tag to compare to (default: last created tag)",
        "files": "Display files instead of commits (default: false)",
        "verbose": "Files: display changes/insertions/deletion."
        " Commits: display the full commit message, author... (default: False)",
        "by_author": "Group commits by author. Doesn't work with --files or --verbose. (default: False)",
    },
)
def changes_since_tag(
    c: Context,
    tag: str = "",
    files: bool = False,
    verbose: bool = False,
    by_author: bool = False,
) -> None:
    """Display changes (commits or files) since the last tag (or a chosen tag)."""
    if files:
        which_tag = tag or run_stdout(c, "git tag --list --sort -creatordate | head -1", hide=False, dry=False)
        default_branch = set_default_branch(c)
        option = "" if verbose else " --name-only"
        c.run(f"git diff --stat {which_tag} origin/{default_branch}{option}")
    else:
        which_tag = tag or "$(git describe --tags --abbrev=0)"
        option = " --format='%aN|%s' | sort -u" if by_author else "" if verbose else " --oneline"
        cmd = f"git log {which_tag}..HEAD{option}"
        if by_author:
            commits_by_author = defaultdict(list)
            for line in run_lines(c, cmd):
                author, commit = line.split("|")
                commits_by_author[author].append(commit)
            for author, commits in commits_by_author.items():
                print(f"\n{author}:")
                for commit in commits:
                    print(f"  {commit}")
        else:
            c.run(cmd)

dirty(c, dir_)

Find Git dirs in multiple directories recursively and print the ones which are dirty.

Source code in src/conjuring/spells/git.py 
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
@task(
    help={
        "dir": "Directory to check recursively for dirty repos. Can be used multiple times. Default: current dir",
    },
    iterable=["dir_"],
    klass=MagicTask,
)
def dirty(c: Context, dir_: list[str | Path]) -> bool:
    """Find Git dirs in multiple directories recursively and print the ones which are dirty."""
    # Use current directory if no directories provided
    if not dir_:
        dir_ = [Path.cwd()]

    # Convert to Path objects and expand user paths
    search_dirs = [Path(d).expanduser().resolve() for d in dir_]

    # Find all Git repositories using fd
    git_repos = _find_git_repositories(c, search_dirs)

    if not git_repos:
        print_warning("No Git repositories found")
        return False

    # Check each repository for dirty status
    dirty_repos_found = False
    for repo_path in sorted(git_repos):
        reasons = _is_repo_dirty(c, repo_path)
        if reasons:
            dirty_repos_found = True

    return dirty_repos_found

extract_subtree(c, new_project_dir, reset=False)

Extract files from subdirectories of the current Git repo to another repo, using git-filter-repo.

This function extracts a subset of files from the current repository's entire Git history into a new repository, preserving all commits that touched those files.

Steps: 1. Prepare new project directory (clone from origin or reset if requested) 2. Gather all files from Git history including deleted files 3. Interactive selection with fzf (TAB to select/deselect) 4. Filter repository history using git-filter-repo with selected paths 5. Display next steps and instructions

Prerequisites: - Install git-filter-repo: pipx install git-filter-repo - Current directory must be a Git repository with a remote origin

Warning: - Rewrites Git history in the new repository - Do NOT push filtered repository to the same remote as the original

Source code in src/conjuring/spells/git.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@task(
    help={
        "new_project_dir": "Dir of the project to be created. The dir might exist or not",
        "reset": "Remove the new dir and start over",
    },
)
def extract_subtree(c: Context, new_project_dir: str, reset: bool = False) -> None:
    """Extract files from subdirectories of the current Git repo to another repo, using git-filter-repo.

    This function extracts a subset of files from the current repository's entire Git history into a new
    repository, preserving all commits that touched those files.

    Steps:
    1. Prepare new project directory (clone from origin or reset if requested)
    2. Gather all files from Git history including deleted files
    3. Interactive selection with fzf (TAB to select/deselect)
    4. Filter repository history using git-filter-repo with selected paths
    5. Display next steps and instructions

    Prerequisites:
    - Install git-filter-repo: `pipx install git-filter-repo`
    - Current directory must be a Git repository with a remote origin

    Warning:
    - Rewrites Git history in the new repository
    - Do NOT push filtered repository to the same remote as the original

    """
    new_project_path = _resolve_repo_path(new_project_dir)
    if reset:
        c.run(f"rm -rf {new_project_path}")

    if not new_project_path.exists():
        origin_url = run_stdout(c, "git remote get-url origin")
        c.run(f"git clone {origin_url} {new_project_path}")

    files_and_dirs = set(run_lines(c, Git.SHOW_ALL_FILE_HISTORY, dry=False))
    for line in sorted(files_and_dirs):
        path = Path(line)
        if os.path.sep in line:
            files_and_dirs.add(str(path.parent) + os.path.sep)
            continue

    _, temp_filename = tempfile.mkstemp()
    temp_file = Path(temp_filename)
    try:
        temp_file.write_text(os.linesep.join(sorted(files_and_dirs)))
        chosen_files = set(
            run_with_fzf(
                c,
                f"cat {temp_filename}",
                dry=False,
                header="Use TAB to choose the files you want to copy to the new project",
                multi=True,
                preview="test -f {} && head -20 {} || echo FILE NOT FOUND, IT EXISTS ONLY IN GIT HISTORY",
            ),
        )
    finally:
        temp_file.unlink()

    with c.cd(new_project_dir):
        all_paths = [f"--path '{line}'" for line in sorted(chosen_files)]
        c.run(f"git-filter-repo {' '.join(all_paths)}")
        history(c, full=True)
    print_error("Don't forget to switch to the new repo:", f"  cd {new_project_dir}", join_nl=True)
    print_success(
        "Next steps:",
        "- Run 'invoke git.rewrite' to fix dates and authors",
        "- Create a new empty repo on https://github.com/new without initializing it (no README/.gitignore/license)",
        "- Follow the instructions to add a remote (from 'push an existing repository from the command line')",
        "- Push files to the new repo with:",
        "  git push -u origin master",
        join_nl=True,
    )

global_config() cached

Global Git configuration.

Source code in src/conjuring/spells/git.py 
45
46
47
48
49
50
@lru_cache
def global_config() -> ConfigParser:
    """Global Git configuration."""
    config = ConfigParser()
    config.read(GLOBAL_GITCONFIG_PATH)
    return config

history(c, full=False, files=False, author=False, dates=False)

Grep the whole Git log and display information.

Source code in src/conjuring/spells/git.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
@task(
    help={
        "full": "Display all info: files, authors, dates",
        "files": "Display all files in Git history, even the ones that were deleted and don't exist anymore",
        "author": "Display authors",
        "dates": "Display committer and author dates in different colors",
    },
)
def history(c: Context, full: bool = False, files: bool = False, author: bool = False, dates: bool = False) -> None:
    """Grep the whole Git log and display information."""
    option_chosen = False
    if full:
        option_chosen = True
        files = author = dates = True
    if files:
        option_chosen = True
        c.run(Git.SHOW_ALL_FILE_HISTORY)
    if author:
        option_chosen = True
        c.run("git log --name-only | rg author | sort -u")
    if dates:
        option_chosen = True
        header = True
        for line in run_lines(c, 'git log --format="%H|%cI|%aI|%GK|%s"', hide=False):
            if header:
                print_success("Green = dates are equal")
                print_error("Red = dates are different")
                typer.echo(
                    "Commit                                   Committer Date            "
                    "Author Date               GPG key          Subject",
                )
                header = False

            fields = line.split("|")
            committer_date = fields[1]
            author_date = fields[2]
            func = print_success if committer_date == author_date else print_error
            func(*fields)
    if not option_chosen:
        msg = "Choose at least one option: --full, --files, --author, --dates"
        vanish(msg, 1)

import_repos(c, target, source='', dir_='', yes=False, rollback=False)

Merge a Git repository into a target repository as a subdirectory, preserving history.

This function safely merges a source repository into a target repository locally, without pushing any changes. All operations are Git-reversible using standard Git commands.

The current branch of the source repository will be used for the merge. If the destination directory is not specified, the source repository name will be used.

Detailed steps: 1. Validate repositories and check that destination directory doesn't exist 2. Create a temporary clone of the source repository 3. Rewrite the temporary clone's history to move all files into the destination subdirectory 4. Add the rewritten repository as a temporary remote and fetch it 5. Merge the unrelated histories into the target repository 6. Clean up temporary remote and clone directory 7. Display rollback instructions

Prerequisites: - Install git-filter-repo: pipx install git-filter-repo - Target repository must exist and be a Git repository - Source repository must exist and be a Git repository

Safety features: - No git push commands - all changes are local only - Source repository is never modified (temporary clone is used) - Target repository can be rolled back using git reset - Creates a tag before merging for easy rollback

Note: git-extras has a git-merge-repo command, but I only found out about it after I had written this function, which does a lot of verifications that git-merge-repo doesn't.

Example usage

invoke git.import-repos --target=/path/to/target --source=/path/to/source --dir_=my-subdir invoke git.import-repos --source=/path/to/source # Uses source repo name as destination dir

Rollback

invoke git.import-repos --target=/path/to/target --rollback

Source code in src/conjuring/spells/git.py 
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
@task(
    help={
        "target": "Path to the target repository where source repo will be merged into",
        "source": "Path to source repository to merge",
        "dir_": "Name of the destination directory in the target repository (default: source repo name)",
        "yes": "Skip confirmation prompt and proceed automatically",
        "rollback": "Rollback a previous import by selecting a safety tag",
    },
)
def import_repos(  # noqa: PLR0913
    c: Context,
    target: str,
    source: str = "",
    dir_: str = "",
    yes: bool = False,
    rollback: bool = False,
) -> None:
    """Merge a Git repository into a target repository as a subdirectory, preserving history.

    This function safely merges a source repository into a target repository locally, without pushing
    any changes. All operations are Git-reversible using standard Git commands.

    The current branch of the source repository will be used for the merge.
    If the destination directory is not specified, the source repository name will be used.

    Detailed steps:
    1. Validate repositories and check that destination directory doesn't exist
    2. Create a temporary clone of the source repository
    3. Rewrite the temporary clone's history to move all files into the destination subdirectory
    4. Add the rewritten repository as a temporary remote and fetch it
    5. Merge the unrelated histories into the target repository
    6. Clean up temporary remote and clone directory
    7. Display rollback instructions

    Prerequisites:
    - Install git-filter-repo: pipx install git-filter-repo
    - Target repository must exist and be a Git repository
    - Source repository must exist and be a Git repository

    Safety features:
    - No git push commands - all changes are local only
    - Source repository is never modified (temporary clone is used)
    - Target repository can be rolled back using git reset
    - Creates a tag before merging for easy rollback

    Note: git-extras has a git-merge-repo command, but I only found out about it after I had written
    this function, which does a lot of verifications that git-merge-repo doesn't.

    Example usage:
      invoke git.import-repos --target=/path/to/target --source=/path/to/source --dir_=my-subdir
      invoke git.import-repos --source=/path/to/source  # Uses source repo name as destination dir

    Rollback:
      invoke git.import-repos --target=/path/to/target --rollback

    """
    if rollback:
        target_path = _resolve_repo_path(target) if target else Path.cwd()
        _handle_rollback(c, target_path)
        return

    if not source:
        vanish("Source repository must be specified using --source")

    target_path = _resolve_repo_path(target) if target else Path.cwd()
    _validate_git_repo(target_path, "Target repository")

    src_path = _resolve_repo_path(source)
    _validate_git_repo(src_path, "Source repository")

    if not dir_:
        dir_ = src_path.name
    dest_dir_path = target_path / dir_
    if dest_dir_path.exists():
        vanish(f"The destination dir {dir_!r} already exists")

    _validate_and_display_repo_status(c, target_path, "Target repository")
    print_success(f"Destination directory: {dir_}")
    src_branch = _validate_and_display_repo_status(c, src_path, "Source repository", show_branches=True)

    if not yes:
        ask_yes_no("Do you want to proceed with merging this repository?")

    timestamp = time.strftime("%Y-%m-%d-%H-%M-%S")
    safety_tag = f"{IMPORT_REPOS_TAG_PREFIX}-{timestamp}"
    tag_message = f"Before merging repo {src_path.name} into {target_path.name}/{dir_}"

    with c.cd(str(target_path)):
        c.run(f"git tag -a {safety_tag} -m '{tag_message}'")
        print_success(f"Created safety tag: {safety_tag}")

    _perform_repo_merge(c, target_path, src_path, src_branch, dir_)

    with c.cd(str(target_path)):
        c.run("git log -1")

    print_success("\nRepository merged successfully!")
    print("\nRollback option if needed:")
    target_flag = f" --target={target_path}" if target else ""
    print(f"  invoke git.import-repos{target_flag} --rollback")
    print("\nOr manually:")
    print(f"  cd {target_path} && git reset --hard {safety_tag}")
    print("\nTo keep the changes, you can delete the safety tag:")
    print(f"  cd {target_path} && git tag -d {safety_tag}")
    print_warning("\nRemember: No changes have been pushed. Run 'git push' when ready.")
    print("\nNote: Source repository was not modified (temporary clone was used).")

is_valid_git_repository(repo_path)

Check if the given path is a valid Git repository.

Source code in src/conjuring/spells/git.py 
464
465
466
467
def is_valid_git_repository(repo_path: Path) -> bool:
    """Check if the given path is a valid Git repository."""
    git_dir: Path = repo_path / DOT_GIT
    return (git_dir / "HEAD").exists() and (git_dir / "refs").exists()

merge_default(c, remote=False, update=True, push=True, rebase=False)

Merge the default branch of the repo. Also set it with "git config", if not already set.

Source code in src/conjuring/spells/git.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
@task(
    help={
        "remote": "List remote branches (default: False)",
        "update": "Update the repo before merging (default: True)",
        "push": "Push the merge to the remote (default: True)",
        "rebase": "Rebase the default branch before merging (default: False)",
    },
)
def merge_default(
    c: Context,
    remote: bool = False,
    update: bool = True,
    push: bool = True,
    rebase: bool = False,
) -> None:
    """Merge the default branch of the repo. Also set it with "git config", if not already set."""
    default_branch = set_default_branch(c, remote)

    if update:
        c.run("git pull")
    which_verb = "rebase" if rebase else "merge"
    run_command(c, f"git {which_verb}", f"origin/{default_branch}")
    if push:
        force_option = "--force-with-lease" if rebase else ""
        run_command(c, "git push", force_option)

new_branch(c, title)

Create a new Git branch with a slugified title while keeping the Jira ticket in uppercase.

Source code in src/conjuring/spells/git.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
@task
def new_branch(c: Context, title: str) -> None:
    """Create a new Git branch with a slugified title while keeping the Jira ticket in uppercase."""
    match = REGEX_JIRA_TICKET_TITLE.match(title)

    if match:
        ticket: str = match.group("ticket")
        title_text: str = match.group("title") or ""
        slugified_title: str = slugify(title_text)
        branch_name: str = f"{ticket}-{slugified_title}" if slugified_title else ticket
    else:
        branch_name = slugify(title)

    typer.echo(f"Creating branch: {branch_name}")
    c.run(f"git checkout -b {branch_name}")

rewrite(c, commit='--root', gpg=True, author=True)

Rewrite a range of commits, signing with GPG and setting the author.

https://git-scm.com/docs/git-commit https://git-scm.com/docs/git-rebase

Source code in src/conjuring/spells/git.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
@task(
    help={
        "commit": "Base commit to be used for the range (default: --root)",
        "gpg": "Sign the commit (default: True)",
        "author": "Set the current author (taken from 'git config') on the commit range",
    },
)
def rewrite(c: Context, commit: str = "--root", gpg: bool = True, author: bool = True) -> None:
    """Rewrite a range of commits, signing with GPG and setting the author.

    https://git-scm.com/docs/git-commit
    https://git-scm.com/docs/git-rebase
    """
    gpg_flag = " --gpg-sign" if gpg else " --no-gpg-sign"

    author_flag = ""
    if author:
        name = run_stdout(c, "git config user.name", dry=False)
        email = run_stdout(c, "git config user.email", dry=False)
        author_flag = f' --author "{name} <{email}>"'

    c.run(f'git log --format="%H %cI %aI %s" {commit} > $TMPDIR/rebase_sign_hashlist')
    c.run(
        "git rebase --committer-date-is-author-date --exec 'GIT_COMMITTER_DATE="
        '$(fgrep -m 1 "$(git log -1 --format="%aI %s" $GIT_COMMIT)" $TMPDIR/rebase_sign_hashlist'
        f' | cut -d" " -f3) git commit --amend --no-edit -n{author_flag}{gpg_flag}\' -i {commit}',
    )
    history(c, dates=True)
    typer.echo()
    typer.echo("NOTE: If commits were modified during the rebase above, their committer date will be the current date")
    typer.echo("Rebase again with this command, without changing any commit, and all dates should be green")

set_default_branch(c, remote=False)

Set the default branch config on the repo, if not configured yet.

Source code in src/conjuring/spells/git.py 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def set_default_branch(c: Context, remote: bool = False) -> str:
    """Set the default branch config on the repo, if not configured yet."""
    cmd_read_default_branch = "git config git-extras.default-branch"
    default_branch = run_stdout(c, cmd_read_default_branch, warn=True, dry=False)
    if not default_branch:
        default_branch = run_with_fzf(
            c,
            "git branch --list",
            "--all" if remote else "",
            "| cut -b 3- | grep -v HEAD | sed -E 's#remotes/[^/]+/##g' | sort -u",
        )
        run_command(c, cmd_read_default_branch, default_branch)
        run_command(c, "git config init.defaultBranch", default_branch)
        run_command(c, "git config --list | rg default.*branch")
    return default_branch

switch_url_to(c, remote='origin', https=False)

Set an SSH or HTTPS URL for a remote.

Source code in src/conjuring/spells/git.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@task
def switch_url_to(c: Context, remote: str = "origin", https: bool = False) -> None:
    """Set an SSH or HTTPS URL for a remote."""
    regex = r"'git@(.+\.com):(.+/.+)\.git\s'" if https else r"'/([^/]+\.com)/([^/]+/.+)\s\('"
    replace = "'$1/$2'" if https else "'$1:$2'"

    result = c.run(f"git remote -v | rg {remote} | head -1 | rg -o {regex} -r {replace}", warn=True, pty=False)
    match = result.stdout.strip()
    if not match:
        typer.echo(f"{Color.BOLD_RED.value}Match not found{Color.NONE.value}")
    else:
        repo = f"https://{match}" if https else f"git@{match}"
        if not repo.endswith(DOT_GIT):
            repo += DOT_GIT
        c.run(f"git remote set-url {remote} {repo}")

    c.run("git remote -v")

tidy_up(c)

Prune remotes, update all branches of the repo, delete merged/squashed branches.

Source code in src/conjuring/spells/git.py 
283
284
285
286
287
288
289
290
291
292
293
@task
def tidy_up(c: Context) -> None:
    """Prune remotes, update all branches of the repo, delete merged/squashed branches."""
    c.run("gitup .")
    c.run("git delete-merged-branches")

    # warn=True is needed; apparently, this command fails when there is no branch, and execution is stopped
    c.run("git delete-squashed-branches", warn=True)

    for remote in run_lines(c, "git remote", dry=False):
        c.run(f"git remote prune {remote}")

update_all(c, group='')

Run gita super to update and clean branches.

Source code in src/conjuring/spells/git.py 
102
103
104
105
106
107
108
109
@task(klass=MagicTask)
def update_all(c: Context, group: str = "") -> None:
    """Run gita super to update and clean branches."""
    parts = ["gita", "super"]
    if group:
        parts.append(group)
    gita_super = " ".join(parts)
    c.run(f"{gita_super} up && {gita_super} delete-merged-branches")

watch(c)

Watch a build on GitHub Actions, then open a pull request or repo after the build is over.

Source code in src/conjuring/spells/git.py 
379
380
381
382
383
384
385
386
387
388
@task()
def watch(c: Context) -> None:
    """Watch a build on GitHub Actions, then open a pull request or repo after the build is over."""
    current_branch = Git(c).current_branch()
    print_success(f"Current branch = {current_branch}")

    c.run("gh run watch", warn=True)
    out = c.run(f"gh pr view {current_branch} --web", warn=True).stdout.strip()
    if "no pull requests found for branch" in out:
        c.run("gh repo view --web")

conjuring.spells.jrnl

Query tags and entries with the jrnl note-taking tool.

edit_last(c, journal='')

Edit the last jrnl entry.

Source code in src/conjuring/spells/jrnl.py 
52
53
54
55
56
57
58
59
@task
def edit_last(c: Context, journal: str = "") -> None:
    """Edit the last jrnl entry."""
    cmd = ["jrnl"]
    if journal:
        cmd.append(journal)
    cmd.append("-1 --edit")
    c.run(" ".join(cmd))

query(c, n=0, contains='', edit=False, fancy=False, short=False, journal='')

Query jrnl entries.

Source code in src/conjuring/spells/jrnl.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@task
def query(  # noqa: PLR0913
    c: Context,
    n: int = 0,
    contains: str = "",
    edit: bool = False,
    fancy: bool = False,
    short: bool = False,
    journal: str = "",
) -> None:
    """Query jrnl entries."""
    format_ = "pretty"
    if fancy:
        format_ = "fancy"
    elif short:
        format_ = "short"

    cmd = ["jrnl"]
    if journal:
        cmd.append(journal)
    if n:
        cmd.append(f"-n {n}")
    cmd.append(f"--format {format_}")
    if contains:
        cmd.append(f"-contains {contains}")
    if edit:
        cmd.append("--edit")
    c.run(" ".join(cmd))

tags(c, sort=False, rg='', journal='')

Query jrnl tags.

Source code in src/conjuring/spells/jrnl.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
@task
def tags(c: Context, sort: bool = False, rg: str = "", journal: str = "") -> None:
    """Query jrnl tags."""
    cmd = ["jrnl"]
    if journal:
        cmd.append(journal)
    cmd.append("--tags")
    if sort:
        cmd.append("| sort -u")
    if rg:
        cmd.append(f"| rg {rg}")
    c.run(" ".join(cmd))

conjuring.spells.k8s

Kubernetes: get pods, show variables from config maps, validate score and more.

Kubectl dataclass

Kubectl commands.

Source code in src/conjuring/spells/k8s.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass
class Kubectl:
    """Kubectl commands."""

    context: Context

    def choose_apps(self, partial_name: str | None = None, *, multi: bool = False) -> list[str]:
        """Select apps from Kubernetes deployments, using a partial app name and fzf.

        Use the current dir as the app name if no partial app name is provided.
        """
        if not partial_name:
            return [Path.cwd().name]

        return cast(
            "list[str]",
            run_with_fzf(
                self.context,
                """kubectl get deployments.apps -o jsonpath='{range .items[*]}{.metadata.name}{"\\n"}{end}'""",
                query=partial_name or "",
                multi=multi,
            ),
        )

    @staticmethod
    def _app_selector(apps: list[str]) -> str:
        """Return the app selector for one or more apps."""
        sorted_unique_apps = sorted(set(apps))
        if len(sorted_unique_apps) == 1:
            return f"-l app={sorted_unique_apps[0]}"
        selector = f" in ({', '.join(sorted_unique_apps)})"
        return f"-l 'app{selector}'"

    def cmd_get(self, resource: str, apps: list[str]) -> str:
        """Return the kubectl get command for one or more apps."""
        return f"kubectl get {resource} {self._app_selector(apps)}"

    def run_get(self, resource: str, apps: list[str]) -> Result:
        """Run the kubectl get command for one or more apps."""
        return run_command(self.context, self.cmd_get(resource, apps))

choose_apps(partial_name=None, *, multi=False)

Select apps from Kubernetes deployments, using a partial app name and fzf.

Use the current dir as the app name if no partial app name is provided.

Source code in src/conjuring/spells/k8s.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def choose_apps(self, partial_name: str | None = None, *, multi: bool = False) -> list[str]:
    """Select apps from Kubernetes deployments, using a partial app name and fzf.

    Use the current dir as the app name if no partial app name is provided.
    """
    if not partial_name:
        return [Path.cwd().name]

    return cast(
        "list[str]",
        run_with_fzf(
            self.context,
            """kubectl get deployments.apps -o jsonpath='{range .items[*]}{.metadata.name}{"\\n"}{end}'""",
            query=partial_name or "",
            multi=multi,
        ),
    )

cmd_get(resource, apps)

Return the kubectl get command for one or more apps.

Source code in src/conjuring/spells/k8s.py 
51
52
53
def cmd_get(self, resource: str, apps: list[str]) -> str:
    """Return the kubectl get command for one or more apps."""
    return f"kubectl get {resource} {self._app_selector(apps)}"

run_get(resource, apps)

Run the kubectl get command for one or more apps.

Source code in src/conjuring/spells/k8s.py 
55
56
57
def run_get(self, resource: str, apps: list[str]) -> Result:
    """Run the kubectl get command for one or more apps."""
    return run_command(self.context, self.cmd_get(resource, apps))

config_map(c, app, rg='')

Show the config map for an app.

Source code in src/conjuring/spells/k8s.py 
69
70
71
72
73
74
75
76
77
78
79
@task(help={"rg": "Filter results with rg"})
def config_map(c: Context, app: str, rg: str = "") -> None:
    """Show the config map for an app."""
    chosen_app = Kubectl(c).choose_apps(app)
    run_command(
        c,
        f"kubectl get deployment/{chosen_app} -o json",
        "| jq -r .spec.template.spec.containers[].envFrom[].configMapRef.name",
        "| rg -v null | xargs -I % kubectl get configmap/% -o json | jq -r .data",
        f"| rg {rg}" if rg else "",
    )

exec_(c, app='')

Exec into the first pod found for the chosen app.

Source code in src/conjuring/spells/k8s.py 
109
110
111
112
113
114
115
116
117
118
119
120
@task(name="exec")
def exec_(c: Context, app: str = "") -> None:
    """Exec into the first pod found for the chosen app."""
    kubectl = Kubectl(c)
    chosen_app = kubectl.choose_apps(app)
    chosen_pod = run_with_fzf(
        c,
        kubectl.cmd_get("pods", chosen_app),
        "--no-headers",
        "-o custom-columns=NAME:.metadata.name",
    )
    run_command(c, f"kubectl exec -it {chosen_pod} -- bash")

pods(c, app='', replica_set=False)

Show the pods and replica sets for an app.

Source code in src/conjuring/spells/k8s.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@task(
    help={
        "app": "Show the pods for an app; if not provided, the current directory name is used.",
        "replica_set": "Show the replica sets for an app",
    },
)
def pods(c: Context, app: str = "", replica_set: bool = False) -> None:
    """Show the pods and replica sets for an app."""
    kubectl = Kubectl(c)
    chosen_apps = kubectl.choose_apps(app, multi=True)
    kubectl.run_get("pods", chosen_apps)

    if replica_set:
        replica_set_names = run_lines(
            c,
            kubectl.cmd_get("pods", chosen_apps),
            """-o jsonpath='{range .items[*]}{.metadata.ownerReferences[0].name}{"\\n"}{end}'""",
            "| sort -u",
        )
        for name in replica_set_names:
            run_command(c, f"kubectl get replicaset {name}")

validate_score(c)

Validate and score files that were changed from the master branch.

Source code in src/conjuring/spells/k8s.py 
60
61
62
63
64
65
66
@task()
def validate_score(c: Context) -> None:
    """Validate and score files that were changed from the master branch."""
    # TODO: handle branches named "main"
    # Continue even if there are errors
    c.run("git diff master.. --name-only | xargs kubeval", warn=True)
    c.run("git diff master.. --name-only | xargs kubectl score")

conjuring.spells.media

Media files: remove empty dirs, clean up picture dirs, download YouTube videos, transcribe audio.

CompareDirsAction

Bases: Enum

Actions to take when comparing two directories.

Source code in src/conjuring/spells/media.py 
165
166
167
168
169
170
171
172
173
class CompareDirsAction(Enum):
    """Actions to take when comparing two directories."""

    # keep-sorted start
    DELETE_IDENTICAL = "identical_deleted"
    DIFF_FAILED = "diff_failed"
    DO_NOTHING = None
    MOVE_IDENTICAL = "identical"
    MOVE_NOT_FOUND = "not_found"

cleanup(c, browse=False)

Cleanup pictures.

Source code in src/conjuring/spells/media.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@task
def cleanup(c: Context, browse: bool = False) -> None:
    """Cleanup pictures."""
    c.run(f"fd -H -0 -tf -i {DOT_DS_STORE} | xargs -0 rm -v")
    c.run(f"fd -H -0 -tf -i {DOT_NOMEDIA} | xargs -0 rm -v")
    c.run("find . -mindepth 1 -type d -empty -print -delete")

    # Unhide Picasa originals dir
    for line in c.run("fd -H -t d .picasaoriginals", pty=False).stdout.splitlines():
        original_dir = Path(line)
        c.run(f"mv {original_dir} {original_dir.parent}/Picasa_Originals")

    # Keep the original dir as the main dir and rename parent dir to "_Copy"
    for line in c.run("fd -t d originals", pty=False).stdout.splitlines():
        original_dir = Path(line)
        c.run(f"mv {original_dir} {original_dir.parent}_Temp")
        c.run(f"mv {original_dir.parent} {original_dir.parent}_Copy")
        c.run(f"mv {original_dir.parent}_Temp {original_dir.parent}")

    # Merge the copy dir with the main one
    for line in run_command(c, "fd -a -uu -t d --color never _copy", str(ONEDRIVE_PICTURES_DIR)).stdout.splitlines():
        copy_dir = Path(line)
        original_dir = Path(line.replace("_Copy", ""))
        if original_dir.exists():
            if browse:
                c.run(f"open '{original_dir}'")
            c.run(f"merge-dirs '{original_dir}' '{copy_dir}'")
        else:
            c.run(f"mv '{copy_dir}' '{original_dir}'")

    # List dirs with _Copy files
    copy_dirs = set()
    for line in run_command(
        c,
        "fd -H -t f --color never _copy",
        str(ONEDRIVE_PICTURES_DIR),
        hide=True,
    ).stdout.splitlines():
        copy_dirs.add(Path(line).parent)

    for dir_ in sorted(copy_dirs):
        typer.echo(dir_)

compare_dirs(c, from_dir, to_dir, count=MAX_COUNT, size=MAX_SIZE, delete=False, move=False, wildcard_search=False)

Compare files in two directories. Stops when it reaches max count or size.

Source code in src/conjuring/spells/media.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
@task(
    help={
        "from_dir": "Source root directory to compare from",
        "to_dir": "Destination root directory to compare to",
        "count": f"Max number of files to compare. Default: {MAX_COUNT}",
        "size": f"Max size of files to compare. Default: {naturalsize(MAX_SIZE)}",
        "delete": "Delete identical files from the source dir",
        "move": "Move identical files from the source dir to the output dir",
        "wildcard_search": "If not found with the exact name,"
        " search the destination dir using the file name with wildcards",
    },
)
def compare_dirs(  # noqa: PLR0913
    c: Context,
    from_dir: str,
    to_dir: str,
    count: int = MAX_COUNT,
    size: int = MAX_SIZE,
    delete: bool = False,
    move: bool = False,
    wildcard_search: bool = False,
) -> None:
    """Compare files in two directories. Stops when it reaches max count or size."""
    if delete and move:
        print_error("Choose either --delete or --move, not both")
        return
    dry = not (delete or move) or c.config.run.dry

    # Use a slug to compare multiple dirs at the same time
    abs_from_dir = Path(from_dir).expanduser().absolute()
    slug = str(abs_from_dir.relative_to(Path.home())).replace(os.sep, "-")
    output_dir = DOWNLOADS_DIR / "compare-dirs-output" / slug

    print_success("Output dir:", str(output_dir), "/ Stop file or dir:", str(STOP_FILE_OR_DIR))

    current_count = 0
    current_size = 0

    max_results = f"--max-results {count}" if count else ""
    lines = run_lines(c, "fd -t f -u", max_results, ".", str(abs_from_dir), "| sort", dry=False)

    with tqdm(lines) as pbar:
        for line in pbar:
            if check_stop_file():
                break

            source_file: Path = Path(line).absolute()
            if source_file.name == DOT_DS_STORE:
                continue

            current_count += 1
            file_size = source_file.stat().st_size
            current_size += file_size

            partial_source_path = source_file.relative_to(abs_from_dir)
            destination_file: Path = to_dir / partial_source_path

            action, file_description = _determine_action(
                c,
                source_file,
                to_dir,
                destination_file,
                delete,
                move,
                wildcard_search,
            )

            pbar.set_postfix(count=current_count, size=naturalsize(file_size), total_size=naturalsize(current_size))

            # Check the file size after running the diff, so remote on-demand files are downloaded locally
            if size and current_size > size:
                print_error(
                    f"Current size ({naturalsize(current_size)})",
                    f"exceeded --size ({naturalsize(size)}), stopping",
                    dry=dry,
                )
                break

            if action == CompareDirsAction.DO_NOTHING:
                print_normal(file_description, dry=dry)
                continue

            _execute(
                action,
                source_file,
                file_description,
                output=output_dir / action.value / partial_source_path,
                dry=dry,
            )

empty_dirs(c, dir_, delete=False, fd=True)

Remove some hidden files first, then remove empty dirs.

The ending slash is needed to search OneDrive, now that its behaviour changed in macOS Monterey.

Source code in src/conjuring/spells/media.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@task(
    help={
        "dir": "Directory to clean up. Default: current dir",
        "fd": "Use https://github.com/sharkdp/fd instead of 'find'",
        "delete": "Delete the actual files (dotfiles are always deleted). Default: False",
    },
    iterable=["dir_"],
)
def empty_dirs(c: Context, dir_: list[str | Path], delete: bool = False, fd: bool = True) -> None:
    """Remove some hidden files first, then remove empty dirs.

    The ending slash is needed to search OneDrive, now that its behaviour changed in macOS Monterey.
    """
    if not dir_:
        dir_ = [Path.cwd()]

    dirs = list({str(Path(d).expanduser().absolute()) for d in dir_})
    xargs = "xargs -0 -r -n 1 rm -v"
    for hidden_file in [DOT_DS_STORE, DOT_NOMEDIA]:
        if fd:
            c.run(f"fd -uu -0 -tf -i {hidden_file} {'/ '.join(dirs)}/ | {xargs}", hide=True)
        else:
            for one_dir in dirs:
                c.run(f"find {one_dir}/ -type f -iname {hidden_file} -print0 | {xargs}", hide=True)

    f_option = " ".join([f"-f {d}/" for d in dirs[:-1]])
    delete_flag = "-delete" if delete else ""
    run_command(c, "find", f_option, f"{dirs[-1]}/ -mindepth 1 -type d -empty -print", delete_flag)
    if not delete:
        print_warning("Run with --delete to actually delete the files", dry=True)

invidious(c)

Parse Invidious instances to be used in the Chrome plugin random instance pool.

"Invidious random instance pool (comma-separated)" text field config from Privacy Redirect.

Source code in src/conjuring/spells/media.py 
408
409
410
411
412
413
414
415
416
417
@task
def invidious(c: Context) -> None:
    """Parse Invidious instances to be used in the Chrome plugin random instance pool.

    "Invidious random instance pool (comma-separated)" text field config
    from [Privacy Redirect](https://chromewebstore.google.com/detail/privacy-redirect/pmcmeagblkinmogikoikkdjiligflglb).
    """
    instances = run_lines(c, "curl -s https://api.invidious.io/instances.json | jq -rs '.[].[][1].uri'")
    filtered = ",".join([i for i in instances if not i.endswith(".onion") and not i.endswith(".i2p")])
    run_command(c, f"echo {filtered} | pbcopy")

slideshow(c, start_at='')

Show pictures in the current dir with feh.

Source code in src/conjuring/spells/media.py 
143
144
145
146
147
@task
def slideshow(c: Context, start_at: str = "") -> None:
    """Show pictures in the current dir with feh."""
    start_at_option = f"--start-at {start_at}" if start_at else ""
    run_command(c, "feh -r -. -g 1790x1070 -B black --caption-path .", start_at_option)

sync_subtitle(c, episode)

Sync subtitles for an episode or movie.

Source code in src/conjuring/spells/media.py 
420
421
422
423
424
425
426
@task
def sync_subtitle(c: Context, episode: str) -> None:
    """Sync subtitles for an episode or movie."""
    fzf_cmd = "fzf --height 40% --reverse --inline-info"
    video = c.run(f"fd {episode} | {fzf_cmd} --header='Choose a video file'", pty=False).stdout.strip()
    subtitle = c.run(f"fd {episode} | {fzf_cmd} --header='Choose a subtitle'", pty=False).stdout.strip()
    c.run(f'ffsubsync --max-offset-seconds 600 --overwrite-input "{video}" -i "{subtitle}"')

unzip_tree(c, dir_, count=1, delete=False)

Unzip .tar.gz files in a directory tree.

Source code in src/conjuring/spells/media.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
@task(
    help={
        "dir": "Root directory to unzip. Default: current dir",
        "count": "Max number of files to unzip. Default: 1",
        "delete": "Delete the .tar.gz file after unzipping with success. Default: False",
    },
    iterable=["dir_"],
)
def unzip_tree(c: Context, dir_: list[str | Path], count: int = 1, delete: bool = False) -> None:
    """Unzip .tar.gz files in a directory tree."""
    if not dir_:
        dir_ = [Path.cwd()]

    for one_dir in dir_:
        for tar_gz_path in iter_path_with_progress(
            c,
            "-t f .tar.gz",
            str(one_dir),
            "| sort --ignore-case",
            max_count=count,
        ):
            result = run_command(c, f"gtar -xzf '{tar_gz_path}' -C '{tar_gz_path.parent}'")
            if result.ok and delete:
                run_command(c, f"rm '{tar_gz_path}'")

whisper(c, dir_)

Transcribe multiple audio file that haven't been transcribed yet, using whisper.

Source code in src/conjuring/spells/media.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
@task(help={"dir_": "Directory with audios to transcribe"})
def whisper(c: Context, dir_: str | Path) -> None:
    """Transcribe multiple audio file that haven't been transcribed yet, using whisper."""
    dir_ = Path(dir_).expanduser()
    audios: list[Path] = []
    for extension in AUDIO_EXTENSIONS:
        audios.extend(dir_.glob(f"*.{extension}"))
    for file in audios:
        transcript_file = file.with_suffix(".txt")
        if not transcript_file.exists():
            c.run(f"whisper --language pt -f txt '{file}' --output_dir '{file.parent}'")
            continue
        c.run(f"open '{transcript_file}'")

youtube_dl(c, url, min_height=360, download_archive_path='')

Download video URLs, try different low-res formats until it finds one.

Source code in src/conjuring/spells/media.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@task
def youtube_dl(c: Context, url: str, min_height: int = 360, download_archive_path: str = "") -> None:
    """Download video URLs, try different low-res formats until it finds one."""
    download_archive_path = download_archive_path or os.environ.get("YOUTUBE_DL_DOWNLOAD_ARCHIVE_PATH", "")
    archive_option = f"--download-archive {download_archive_path!r}" if download_archive_path else ""

    all_heights = [h for h in [240, 360, 480, 0] if h >= min_height or h == 0]
    for height in all_heights:
        # https://github.com/ytdl-org/youtube-dl#format-selection-examples
        # Download best format available but no better than the chosen height
        fmt = f"-f 'bestvideo[height<={height}]+bestaudio/best[height<={height}]'" if height else ""

        result = run_command(
            c,
            "yt-dlp --ignore-errors --restrict-filenames",
            # "--get-title --get-id",
            # "--get-thumbnail --get-description --get-duration --get-filename",
            # "--get-format",
            archive_option,
            fmt,
            url,
            warn=True,
        )
        if result.ok or "Unsupported URL:" in result.stdout:
            break

zip_tree(c, dir_, count=1, depth=5, delete=False)

Zip files in a directory tree, creating a .tar.gz file.

Source code in src/conjuring/spells/media.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
@task(
    help={
        "dir": "Root directory to zip. Default: current dir",
        "count": "Max number of sub dirs to zip. Default: 1",
        "delete": "Delete the directory after zipping with success. Default: False",
    },
    iterable=["dir_"],
)
def zip_tree(c: Context, dir_: list[str | Path], count: int = 1, depth: int = 5, delete: bool = False) -> None:
    """Zip files in a directory tree, creating a .tar.gz file."""
    if not dir_:
        dir_ = [Path.cwd()]

    for raw_dir in dir_:
        path_dir = Path(raw_dir)
        for path_to_zip in iter_path_with_progress(
            c,
            "-t f --exclude '*.tar.gz' .",
            str(raw_dir),
            "--exec echo {//} | sort --unique --ignore-case",
            max_count=count,
            reverse_depth=depth,
        ):
            if path_to_zip == path_dir:
                continue

            tar_gz_file = unique_file_name(path_to_zip.with_suffix(".tar.gz"))
            with c.cd(path_to_zip.parent):
                run_command(
                    c,
                    "gtar",
                    "--remove-files" if delete else "",
                    "--exclude='*.tar.gz'",
                    f'-czf "{tar_gz_file.name}" -C . "./{path_to_zip.name}"',
                    warn=True,
                )

conjuring.spells.mkdocs

MkDocs spells: install, build, deploy to GitHub, serve locally.

browse(c)

Open the static HTML docs website on your browser.

Source code in src/conjuring/spells/mkdocs.py 
62
63
64
65
@task
def browse(c: Context) -> None:
    """Open the static HTML docs website on your browser."""
    c.run("open http://127.0.0.1:8000/")

build(c)

Build docs.

Source code in src/conjuring/spells/mkdocs.py 
43
44
45
46
@task
def build(c: Context) -> None:
    """Build docs."""
    c.run("mkdocs build --strict --verbose --site-dir site")

deploy(c)

Deploy docs to GitHub pages.

Source code in src/conjuring/spells/mkdocs.py 
49
50
51
52
@task(pre=[build])
def deploy(c: Context) -> None:
    """Deploy docs to GitHub pages."""
    c.run("mkdocs gh-deploy")

install(c, force=False)

Install MkDocs globally with the Material plugin. Upgrade if it already exists.

Source code in src/conjuring/spells/mkdocs.py 
24
25
26
27
28
29
30
31
32
33
34
@task(help={"force": "Force re-installation of MkDocs."})
def install(c: Context, force: bool = False) -> None:
    """Install MkDocs globally with the Material plugin. Upgrade if it already exists."""
    upgrade = " || pipx upgrade mkdocs" if force else ""
    c.run(f"pipx install mkdocs{upgrade}", warn=True)
    for extension in EXTENSIONS:
        c.run(f"pipx inject mkdocs {extension}")

    # Inject the local project into the global MkDocs installation.
    if has_pyproject_toml():
        c.run("pipx inject mkdocs -e .")

serve(c, browse=True)

Start the live-reloading server to test the docs locally. The "serve" command already builds the docs.

Source code in src/conjuring/spells/mkdocs.py 
55
56
57
58
59
@task
def serve(c: Context, browse: bool = True) -> None:
    """Start the live-reloading server to test the docs locally. The "serve" command already builds the docs."""
    opt_open = "--open" if browse else ""
    run_command(c, "mkdocs serve --livereload --watch-theme", opt_open)

uninstall(c)

Uninstall MkDocs globally.

Source code in src/conjuring/spells/mkdocs.py 
37
38
39
40
@task
def uninstall(c: Context) -> None:
    """Uninstall MkDocs globally."""
    c.run("pipx uninstall mkdocs")

conjuring.spells.mr

myrepos repository management tool: grep text in repos.

MyRepos dataclass

Find and interact with myrepos config files.

Source code in src/conjuring/spells/mr.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@dataclass
class MyRepos:
    """Find and interact with myrepos config files."""

    context: Context

    def find_configs(self, partial_name: str, echo: bool = False) -> list[Path]:
        """Find config files in the current dir or dirs above."""
        lower_partial_name = partial_name.lower()
        glob_pattern = MRCONFIG_FILE if not lower_partial_name else f"{MRCONFIG_FILE}*{lower_partial_name}*"
        config_dir = self._find_dir_with_mrconfigs(glob_pattern)
        if not config_dir:
            msg = f"No {MRCONFIG_FILE}* file was found in {Path.cwd()} or its parents"
            raise FileNotFoundError(msg)

        if not lower_partial_name:
            return [config_dir / MRCONFIG_FILE]

        with self.context.cd(str(config_dir)):
            chosen = run_with_fzf(
                self.context,
                "ls -1",
                f"{MRCONFIG_FILE}*",
                query=lower_partial_name,
                multi=True,
                echo=echo,
                hide=not echo,
            )
        return sorted({config_dir / c for c in chosen})

    @staticmethod
    def _find_dir_with_mrconfigs(glob_pattern: str) -> Path | None:
        for dir_ in chain([Path.cwd()], Path.cwd().parents):
            for _ in dir_.glob(glob_pattern):
                # Exit loop on the first file found; fzf will handle the rest
                return dir_
        return None

find_configs(partial_name, echo=False)

Find config files in the current dir or dirs above.

Source code in src/conjuring/spells/mr.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def find_configs(self, partial_name: str, echo: bool = False) -> list[Path]:
    """Find config files in the current dir or dirs above."""
    lower_partial_name = partial_name.lower()
    glob_pattern = MRCONFIG_FILE if not lower_partial_name else f"{MRCONFIG_FILE}*{lower_partial_name}*"
    config_dir = self._find_dir_with_mrconfigs(glob_pattern)
    if not config_dir:
        msg = f"No {MRCONFIG_FILE}* file was found in {Path.cwd()} or its parents"
        raise FileNotFoundError(msg)

    if not lower_partial_name:
        return [config_dir / MRCONFIG_FILE]

    with self.context.cd(str(config_dir)):
        chosen = run_with_fzf(
            self.context,
            "ls -1",
            f"{MRCONFIG_FILE}*",
            query=lower_partial_name,
            multi=True,
            echo=echo,
            hide=not echo,
        )
    return sorted({config_dir / c for c in chosen})

grep(c, search_text, config='', echo=False)

Grep mr repositories with a search text and print the directories in which the text was found.

Needs mr to be preconfigured with files starting with the ".mrconfig" prefix.

Source code in src/conjuring/spells/mr.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@task(
    help={
        "config": f"Specific config file to use. Use fzf if multiple are found. Default: {MRCONFIG_FILE}",
        "echo": "Echo the commands being executed, for debugging purposes. Default: False",
    },
)
def grep(c: Context, search_text: str, config: str = "", echo: bool = False) -> None:
    """Grep mr repositories with a search text and print the directories in which the text was found.

    Needs mr to be preconfigured with files starting with the ".mrconfig" prefix.
    """
    for chosen in MyRepos(c).find_configs(config, echo=echo):
        # For some reason, using run_command() prints a "\r" char at the end of each line;
        # the solution is to get output as a string and use print().
        output_without_linefeed = run_stdout(
            c,
            "mr -c",
            str(chosen),
            "-m grep",
            search_text,
            "| rg --color=never 'mr grep: (.+)$' --replace '$1'",
            echo=echo,
        )
        typer.echo(output_without_linefeed)

conjuring.spells.onedrive

OneDrive: list files with conflicts.

conflicts(c, dir_)

List files with conflicts.

Source code in src/conjuring/spells/onedrive.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@task(
    help={"dir": "Directory; can be used multiple times. Default: current dir"},
    iterable=["dir_"],
)
def conflicts(c: Context, dir_: list[str | Path]) -> None:
    """List files with conflicts."""
    if not dir_:
        dir_ = [Path.cwd()]

    hostname = run_stdout(c, "hostname -s").strip()
    suffix = f"-{hostname}"
    for one_dir in list({str(Path(d).expanduser().absolute()) for d in dir_}):
        for line in run_lines(c, f"fd -t f {hostname} {one_dir} | sort"):
            duplicated = Path(line)
            original_name = duplicated.stem[: -len(suffix)]
            original = duplicated.with_stem(original_name)
            typer.echo(run_stdout(c, f"diff {duplicated} {original}", warn=True).strip())

force_downloads(c, dir_)

Force downloads of remote OneDrive files by reading them with a dummy "diff".

Source code in src/conjuring/spells/onedrive.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
@task(
    help={"dir": "Directory; can be used multiple times. Default: current dir"},
    iterable=["dir_"],
)
def force_downloads(c: Context, dir_: list[str | Path]) -> None:
    """Force downloads of remote OneDrive files by reading them with a dummy "diff"."""
    if not dir_:
        dir_ = [Path.cwd()]
    temp_file = Path().home() / "delete-me"
    temp_file.touch(exist_ok=True)
    for one_dir in dir_:
        c.run(f"fd -t f -0 . {one_dir} | sort -z | xargs -0 -n 1 diff {temp_file}")
    temp_file.unlink()

conjuring.spells.paperless

Paperless: maintenance, renamer, sanity, delete duplicates.

Document dataclass

A paperless document.

Source code in src/conjuring/spells/paperless.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@dataclass
class Document:
    """A paperless document."""

    document_id: int
    title: str
    errors: list = field(default_factory=list, init=False)

    @property
    def url(self) -> str:
        """Return the URL to the document details page."""
        return f"{paperless_url()}/documents/{self.document_id}/details"

    def __repr__(self) -> str:
        """Return a string representation of the document."""
        return f"Document(document_id={self.document_id}, title='{self.title}', url='{self.url}', errors={self.errors})"

url property

Return the URL to the document details page.

__repr__()

Return a string representation of the document.

Source code in src/conjuring/spells/paperless.py 
115
116
117
def __repr__(self) -> str:
    """Return a string representation of the document."""
    return f"Document(document_id={self.document_id}, title='{self.title}', url='{self.url}', errors={self.errors})"

OrphanFile dataclass

A paperless orphan file.

Source code in src/conjuring/spells/paperless.py 
120
121
122
123
124
125
126
127
128
@dataclass
class OrphanFile:
    """A paperless orphan file."""

    source: Path
    destination: Path

    def __lt__(self, other: OrphanFile) -> bool:
        return self.source < other.source

delete_failed_duplicates(c, max_delete=100)

Delete records marked as duplicate but that cannot be downloaded. So the PDF files can be reimported.

Source code in src/conjuring/spells/paperless.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
@task
def delete_failed_duplicates(c: Context, max_delete: int = 100) -> None:
    """Delete records marked as duplicate but that cannot be downloaded. So the PDF files can be reimported."""
    session = requests.Session()
    session.headers.update({"authorization": f"token {paperless_token()}"})

    delete_count = 0
    req_tasks = session.get(f"{paperless_url()}/api/tasks/?format=json")
    for obj in req_tasks.json():
        if obj["status"] != "FAILURE":
            continue

        raw_line = obj["result"]
        if DUPLICATE_OF not in raw_line:
            print_error(f"Unknown error: {raw_line}")
            continue

        clean_line = raw_line.replace(" Not consuming ", "").replace(DUPLICATE_OF, "")
        first, second, duplicate_with_id = clean_line.split(":", maxsplit=2)
        if first != second:
            print_error(f"Files are different: {first=} / {second=}")
        match = REGEX_TITLE_WITH_ID.match(duplicate_with_id)
        if not match:
            print_error(f"Line doesn't match regex {duplicate_with_id=}", clean_line)
            continue

        data = match.groupdict()
        document_id = data["id"]

        api_document_url = f"{paperless_url()}/api/documents/{document_id}/"
        document_url = f"{paperless_url()}/documents/{document_id}"
        url = f"{api_document_url}download/"
        req_download = session.head(url)
        if req_download.status_code != HTTPStatus.NOT_FOUND:
            print_success(document_url, f"Document exists {req_download.status_code=}", clean_line)
            continue

        req_document = session.head(api_document_url)
        if req_document.status_code == HTTPStatus.NOT_FOUND:
            print_warning(document_url, "Document already deleted before", clean_line)
            continue

        req_delete = session.delete(api_document_url)
        if req_delete.status_code == HTTPStatus.NO_CONTENT:
            print_success(document_url, f"Document deleted #{delete_count}", clean_line)
            delete_count += 1
            if delete_count >= max_delete:
                raise SystemExit
            continue

        print_error(document_url, clean_line, f"Something wrong: {req_delete.status_code=}")
        c.run(f"open {document_url}")

empty_consume_dir(c, instance)

Empty the consume dir.

Source code in src/conjuring/spells/paperless.py 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
@task
def empty_consume_dir(c: Context, instance: str) -> None:
    """Empty the consume dir."""
    consume_dir = paperless_root_dir(instance) / "consume-into-paperless"
    print_success(str(consume_dir))

    fd_cmd = f"fd -uu -t f . {consume_dir}"
    if not c.run(fd_cmd).stdout:
        print_success("Consume dir is already empty.")
        return

    if ask_user_prompt("Delete files above?", allowed_keys="yn") != "y":
        raise SystemExit

    c.run(fd_cmd + " -0 | xargs -0 -r -n 1 rm -v")

    media.empty_dirs(c, [consume_dir], delete=True)
    c.run(f"ls -l {consume_dir}")

maintenance(c, instance, reindex=True, optimize=True, thumbnails=True)

Reindex all docs and optionally optimize them.

https://docs.paperless-ngx.com/administration/#index https://docs.paperless-ngx.com/administration/#thumbnails

Source code in src/conjuring/spells/paperless.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@task
def maintenance(
    c: Context,
    instance: str,
    reindex: bool = True,
    optimize: bool = True,
    thumbnails: bool = True,
) -> None:
    """Reindex all docs and optionally optimize them.

    https://docs.paperless-ngx.com/administration/#index
    https://docs.paperless-ngx.com/administration/#thumbnails
    """
    if reindex:
        c.run(f"{paperless_cmd(instance)} document_index reindex")
    if optimize:
        c.run(f"{paperless_cmd(instance)} document_index optimize")
    if thumbnails:
        c.run(f"{paperless_cmd(instance)} document_thumbnails")

paperless_cmd(instance='')

Command to run Paperless with Docker.

Source code in src/conjuring/spells/paperless.py 
39
40
41
42
43
44
45
46
47
48
def paperless_cmd(instance: str = "") -> str:
    """Command to run Paperless with Docker."""
    value = lazy_env_variable("PAPERLESS_COMPOSE_YAML", "path to the Paperless Docker compose YAML file")
    yaml_file = Path(value).expanduser()
    if not yaml_file.exists():
        msg = f"Paperless compose YAML file doesn't exist: {yaml_file}"
        raise FileNotFoundError(msg)

    suffix = f"_{instance}" if instance else ""
    return f"docker compose -f {yaml_file} exec webserver{suffix}"

paperless_root_dir(instance='')

Root directory for Paperless.

Source code in src/conjuring/spells/paperless.py 
51
52
53
54
55
56
57
58
59
def paperless_root_dir(instance: str = "") -> Path:
    """Root directory for Paperless."""
    value = lazy_env_variable("PAPERLESS_ROOT_DIR", "root directory for Paperless")
    subdir = "paperless_" + instance
    root_dir = Path(value).expanduser() / subdir
    if not root_dir.exists():
        msg = f"Paperless root directory doesn't exist: {root_dir}"
        raise FileNotFoundError(msg)
    return root_dir

paperless_token()

Auth token to access Paperless API.

Source code in src/conjuring/spells/paperless.py 
67
68
69
def paperless_token() -> str:
    """Auth token to access Paperless API."""
    return lazy_env_variable("PAPERLESS_TOKEN", "auth token to access Paperless API")

paperless_url()

URL where Paperless is running.

Source code in src/conjuring/spells/paperless.py 
62
63
64
def paperless_url() -> str:
    """URL where Paperless is running."""
    return lazy_env_variable("PAPERLESS_URL", "URL where Paperless is running")

rename(c, instance)

Rename files.

https://docs.paperless-ngx.com/administration/#renamer

Source code in src/conjuring/spells/paperless.py 
93
94
95
96
97
98
99
@task
def rename(c: Context, instance: str) -> None:
    """Rename files.

    https://docs.paperless-ngx.com/administration/#renamer
    """
    c.run(f"{paperless_cmd(instance)} document_renamer")

sanity(c, instance, hide=False, orphans=False, thumbnails=False, documents=False, unknown=True, errors='', together=False, fix=False, move=False)

Sanity checker. Optionally fix orphan files (copies or movies them to the download dir).

https://docs.paperless-ngx.com/administration/#sanity-checker

Source code in src/conjuring/spells/paperless.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
@task(
    help={
        "hide": "Hide progress bar of sanity command",
        "orphans": "Show orphan files",
        "thumbnails": "Show thumbnail files",
        "documents": "Show documents with issues",
        "unknown": "Show unknown lines from the log",
        "errors": "Show documents with specific errors (comma-separated list)",
        "together": f"Keep {ORPHAN_ORIGINALS} and {ORPHAN_ARCHIVE} in the same output directory",
        "fix": "Fix broken files by copying/moving them to the downloads dir",
        "move": "Move files instead of copying",
    },
)
def sanity(  # noqa: PLR0913
    c: Context,
    instance: str,
    hide: bool = False,
    orphans: bool = False,
    thumbnails: bool = False,
    documents: bool = False,
    unknown: bool = True,
    errors: str = "",
    together: bool = False,
    fix: bool = False,
    move: bool = False,
) -> None:
    """Sanity checker. Optionally fix orphan files (copies or movies them to the download dir).

    https://docs.paperless-ngx.com/administration/#sanity-checker
    """
    # Fail fast if the env var is not set
    documents_dir = paperless_root_dir(instance) / "media/documents"
    if documents_dir and not documents_dir.exists():
        msg = f"Documents directory doesn't exist: {documents_dir}"
        raise RuntimeError(msg)

    # TODO: fix(paperless): implement dry-run mode with dry=False and actually avoid files being copied/moved
    lines = run_lines(c, paperless_cmd(instance), "document_sanity_checker", hide=hide, warn=True, pty=True)

    original_or_archive_files: dict[str, list[OrphanFile]] = defaultdict(list)
    matched_files: list[OrphanFile] = []
    unmatched_files: list[OrphanFile] = []
    orphan_files: list[str] = []
    thumbnail_files: list[str] = []
    current_document: Document | None = None
    documents_with_issues: list[Document] = []
    unknown_lines = []
    for line in lines:
        if "it/s]" in line:
            continue

        msg = "Orphaned file in media dir: "
        if msg in line:
            partial_path = Path(line.split(msg)[1].replace(USR_SRC_DOCUMENTS, ""))
            _process_orphans(partial_path, documents_dir, original_or_archive_files, orphan_files, thumbnail_files)
            continue

        msg = "Detected following issue(s) with document #"
        if msg in line:
            # Append the previous document
            if current_document:
                documents_with_issues.append(current_document)

            # Extract document ID and title from the line
            # Format: "Detected following issue(s) with document #123, titled Title"
            rest_of_line = line.split(msg)[1]
            document_id, title = rest_of_line.split(", titled ")

            current_document = Document(int(document_id), title)
            continue

        if current_document:
            _, error = line.split("[paperless.sanity_checker] ")
            current_document.errors.append(error)
            continue

        unknown_lines.append(line)

    _split_matched_unmatched(original_or_archive_files, matched_files, unmatched_files, together)

    _handle_items(fix, move, orphans, "Matched originals/archive pairs", matched_files)
    _handle_items(fix, move, orphans, "Unmatched originals/archive files", unmatched_files)
    _handle_items(False, move, orphans, "Other orphaned files", orphan_files)
    # TODO: feat(paperless): move thumbnail files to downloads dir
    _handle_items(fix, move, thumbnails, "Orphaned thumbnail files", thumbnail_files)
    _handle_items(False, move, documents, "Documents with issues", documents_with_issues, errors)
    _handle_items(False, move, unknown, "Unknown lines", unknown_lines)

conjuring.spells.pre_commit

pre-commit: install, uninstall, run/autoupdate selected hooks.

auto(c, repo='', bleed=False)

Autoupdate a Git hook or all hooks with the latest tag. Needs fzf and yq.

Source code in src/conjuring/spells/pre_commit.py 
119
120
121
122
123
124
125
126
@task()
def auto(c: Context, repo: str = "", bleed: bool = False) -> None:
    """Autoupdate a Git hook or all hooks with the latest tag. Needs fzf and yq."""
    command = ""
    if repo:
        chosen = run_with_fzf(c, "yq e '.repos[].repo' .pre-commit-config.yaml", query=repo, dry=False)
        command = f"--repo {chosen}"
    run_command(c, "prek auto-update", "--bleeding-edge" if bleed else "", command)

get_hook_types(commit_msg, desired_hooks=None)

Prepare a list of hook types to install/uninstall.

Source code in src/conjuring/spells/pre_commit.py 
48
49
50
51
52
53
54
55
56
def get_hook_types(commit_msg: bool, desired_hooks: list[str] | None = None) -> str:
    """Prepare a list of hook types to install/uninstall."""
    hooks = ["pre-commit"]
    if desired_hooks:
        hooks.extend(desired_hooks)
    if commit_msg:
        hooks.append("commit-msg")
        hooks.append("prepare-commit-msg")
    return " ".join([f"--hook-type {h}" for h in hooks])

install(c, before, gc=False, commit_msg=True)

Install pre-commit with prek.

Source code in src/conjuring/spells/pre_commit.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@task(
    help={
        "gc": "Run the garbage collector to remove unused venvs",
        "commit_msg": "Install commit message hooks",
        "before": "Config files to run before the current one.",
    },
    iterable=["before"],
)
def install(c: Context, before: list[str], gc: bool = False, commit_msg: bool = True) -> None:
    """Install pre-commit with prek."""
    if gc:
        _run_garbage_collector(c)
    c.run(f"prek install {get_hook_types(commit_msg)} --install-hooks")
    if before:
        _patch_pre_commit_configs(before)

run(c, hooks)

Run all pre-commit hooks or a specific one using prek. Don't stop on failures. Needs fzf and yq.

Source code in src/conjuring/spells/pre_commit.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@task(
    help={
        "hooks": "Comma-separated list of partial hook IDs (fzf will be used to match them)."
        " Use 'all', '.' or '-' to run all hooks.",
    },
)
def run(c: Context, hooks: str) -> None:
    """Run all pre-commit hooks or a specific one using prek. Don't stop on failures. Needs fzf and yq."""
    split_hooks = hooks.split(",")
    chosen_hooks = []
    for special in ("all", ".", "-"):
        if special in split_hooks:
            chosen_hooks.append("")
            break
    if not chosen_hooks:
        chosen_hooks = [
            run_with_fzf(
                c,
                "yq e '.repos[].hooks[].id' .pre-commit-config.yaml | sort -u",
                query=partial_hook,
                dry=False,
            )
            for partial_hook in split_hooks
        ]

    for chosen_hook in chosen_hooks:
        run_command(c, "prek run --all-files", chosen_hook, warn=True)

uninstall(c, gc=False, commit_msg=True)

Uninstall ALL pre-commit hooks with prek.

Source code in src/conjuring/spells/pre_commit.py 
76
77
78
79
80
81
82
83
84
85
86
87
@task(help={"gc": "Run the garbage collector to remove unused venvs", "commit_msg": "Uninstall commit message hooks"})
def uninstall(c: Context, gc: bool = False, commit_msg: bool = True) -> None:
    """Uninstall ALL pre-commit hooks with prek."""
    if gc:
        _run_garbage_collector(c)

    installed_hooks = [
        git_hook
        for git_hook in run_stdout(c, "ls .git/hooks", dry=False).splitlines()
        if ".sample" not in git_hook and ".legacy" not in git_hook
    ]
    c.run(f"prek uninstall {get_hook_types(commit_msg, installed_hooks)}")

conjuring.spells.py

Python and Poetry.

Install venvs, run tests and coverage, install debug tools.

Poetry dataclass

Poetry-related tasks.

Source code in src/conjuring/spells/py.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@dataclass()
class Poetry:
    """Poetry-related tasks."""

    context: Context

    def used_in_project(self, display_error: bool = True) -> bool:
        """Check if Poetry is being used."""
        used = int(
            run_command(
                self.context,
                f"grep tool.poetry {PYPROJECT_TOML} 2>/dev/null | wc -c",
                hide=True,
                warn=True,
            ).stdout.strip(),
        )
        if not used and display_error:
            print_error("This task only works with Poetry projects (so far).")
        return bool(used)

    @staticmethod
    def parse_python_version(venv: str) -> str:
        """For now, assuming we only have Poetry venvs."""
        return venv.split(" ")[0].split("-py")[1]

    def remove_venv(self, python_version: str) -> Result:
        """Remove a Poetry venv."""
        return self.context.run(f"poetry env remove python{python_version}")

    def guess_python_version(self) -> str:
        """Guess Python version from pyproject.toml."""
        # TODO: rewrite this hack and use a TOML package to read the values directly
        pyproject_lines = run_lines(
            self.context,
            f"rg --no-line-number -e '^python ' -e python_version {PYPROJECT_TOML}",
        )
        versions: set[str] = set()
        for line in pyproject_lines:
            value_with_comment = line.split("=")[1]
            value_only = value_with_comment.split("#")[0]
            clean_version = value_only.replace("^", "").replace("~", "").strip('" ')
            versions.add(clean_version)
        if len(versions) > 1:
            print_error(f"Multiple Python versions found in {PYPROJECT_TOML}: {versions=}")
            raise SystemExit
        return next(iter(versions))

    def use_venv(self, python_version: str) -> Result:
        """Use a Poetry venv."""
        version_obj = parse(python_version)
        return self.context.run(f"poetry env use python{version_obj.major}.{version_obj.minor}")

guess_python_version()

Guess Python version from pyproject.toml.

Source code in src/conjuring/spells/py.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def guess_python_version(self) -> str:
    """Guess Python version from pyproject.toml."""
    # TODO: rewrite this hack and use a TOML package to read the values directly
    pyproject_lines = run_lines(
        self.context,
        f"rg --no-line-number -e '^python ' -e python_version {PYPROJECT_TOML}",
    )
    versions: set[str] = set()
    for line in pyproject_lines:
        value_with_comment = line.split("=")[1]
        value_only = value_with_comment.split("#")[0]
        clean_version = value_only.replace("^", "").replace("~", "").strip('" ')
        versions.add(clean_version)
    if len(versions) > 1:
        print_error(f"Multiple Python versions found in {PYPROJECT_TOML}: {versions=}")
        raise SystemExit
    return next(iter(versions))

parse_python_version(venv) staticmethod

For now, assuming we only have Poetry venvs.

Source code in src/conjuring/spells/py.py 
67
68
69
70
@staticmethod
def parse_python_version(venv: str) -> str:
    """For now, assuming we only have Poetry venvs."""
    return venv.split(" ")[0].split("-py")[1]

remove_venv(python_version)

Remove a Poetry venv.

Source code in src/conjuring/spells/py.py 
72
73
74
def remove_venv(self, python_version: str) -> Result:
    """Remove a Poetry venv."""
    return self.context.run(f"poetry env remove python{python_version}")

use_venv(python_version)

Use a Poetry venv.

Source code in src/conjuring/spells/py.py 
94
95
96
97
def use_venv(self, python_version: str) -> Result:
    """Use a Poetry venv."""
    version_obj = parse(python_version)
    return self.context.run(f"poetry env use python{version_obj.major}.{version_obj.minor}")

used_in_project(display_error=True)

Check if Poetry is being used.

Source code in src/conjuring/spells/py.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def used_in_project(self, display_error: bool = True) -> bool:
    """Check if Poetry is being used."""
    used = int(
        run_command(
            self.context,
            f"grep tool.poetry {PYPROJECT_TOML} 2>/dev/null | wc -c",
            hide=True,
            warn=True,
        ).stdout.strip(),
    )
    if not used and display_error:
        print_error("This task only works with Poetry projects (so far).")
    return bool(used)

PyEnv dataclass

pyenv-related tasks.

Source code in src/conjuring/spells/py.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@dataclass
class PyEnv:
    """pyenv-related tasks."""

    context: Context

    def has_local(self) -> bool:
        """Check if a local Python version is set."""
        output = self.context.run("pyenv local", warn=True).stdout.strip()
        return output and "no local version" not in output

    def set_local(self, python_version: str) -> Result:
        """Set the local pyenv version."""
        latest = self.list_versions(python_version)[-1]
        return self.context.run(f"pyenv local {latest}")

    def list_versions(self, python_version: str | None = None) -> list[str]:
        """List all installed Python versions, or only the ones matching the desired version."""
        all_versions = run_lines(self.context, "pyenv versions --bare")
        if not python_version:
            return all_versions

        return [version for version in all_versions if version.startswith(python_version)]

has_local()

Check if a local Python version is set.

Source code in src/conjuring/spells/py.py 
28
29
30
31
def has_local(self) -> bool:
    """Check if a local Python version is set."""
    output = self.context.run("pyenv local", warn=True).stdout.strip()
    return output and "no local version" not in output

list_versions(python_version=None)

List all installed Python versions, or only the ones matching the desired version.

Source code in src/conjuring/spells/py.py 
38
39
40
41
42
43
44
def list_versions(self, python_version: str | None = None) -> list[str]:
    """List all installed Python versions, or only the ones matching the desired version."""
    all_versions = run_lines(self.context, "pyenv versions --bare")
    if not python_version:
        return all_versions

    return [version for version in all_versions if version.startswith(python_version)]

set_local(python_version)

Set the local pyenv version.

Source code in src/conjuring/spells/py.py 
33
34
35
36
def set_local(self, python_version: str) -> Result:
    """Set the local pyenv version."""
    latest = self.list_versions(python_version)[-1]
    return self.context.run(f"pyenv local {latest}")

Pytest

Pytest-related tasks.

Source code in src/conjuring/spells/py.py 
100
101
102
103
104
105
106
107
108
109
class Pytest:
    """Pytest-related tasks."""

    @staticmethod
    def command(s: bool) -> str:
        """Build pytest command."""
        command = "pytest -v"
        if s:
            command += " -s"
        return command

command(s) staticmethod

Build pytest command.

Source code in src/conjuring/spells/py.py 
103
104
105
106
107
108
109
@staticmethod
def command(s: bool) -> str:
    """Build pytest command."""
    command = "pytest -v"
    if s:
        command += " -s"
    return command

coverage(c, show_all=False, s=False)

Run tests with pytest and coverage.

Source code in src/conjuring/spells/py.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
@task(
    help={
        "show_all": "Show all lines, even if they are covered",
        "s": "Don't capture output (same shortcut as pytest)",
    },
)
def coverage(c: Context, show_all: bool = False, s: bool = False) -> None:
    """Run tests with pytest and coverage."""
    if not Poetry(c).used_in_project():
        return

    options = [f"--cov={source}" for source in ["src", Path.cwd().name, "app"] if Path(source).exists()]

    skip_option = "" if show_all else ":skip-covered"
    options.append(f"--cov-report=term-missing{skip_option}")

    run_command(c, "poetry run", Pytest.command(s), *options)

debug_tools(c, all_=False, ipython=False, ipdb=False, pudb=False, icecream=False, devtools=False, watch=False, watcher=False)

Install debug tools.

Source code in src/conjuring/spells/py.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
@task(
    help={
        "all_": "Install all debug tools",
        "ipython": "Install https://pypi.org/project/ipython/",
        "ipdb": "Install https://pypi.org/project/ipdb/",
        "pudb": "Install https://pypi.org/project/pudb/",
        "icecream": "Install https://pypi.org/project/icecream/",
        "devtools": "Install https://pypi.org/project/devtools/",
        "watch": "Install https://github.com/joeyespo/pytest-watch",
        "watcher": "Install https://github.com/olzhasar/pytest-watcher",
    },
)
def debug_tools(  # noqa: PLR0913
    c: Context,
    all_: bool = False,
    ipython: bool = False,
    ipdb: bool = False,
    pudb: bool = False,
    icecream: bool = False,
    devtools: bool = False,
    watch: bool = False,
    watcher: bool = False,
) -> None:
    """Install debug tools."""
    if not Poetry(c).used_in_project():
        return

    if watch and watcher:
        print_error("Use only one of --watch and --watcher.")
        return
    if watch:
        c.run("poetry run python -m pip uninstall -y pytest-watcher")
    elif watcher:
        c.run("poetry run python -m pip uninstall -y pytest-watch")
    tools = [
        "pip",
        "ipython" if ipython or all_ else "",
        "ipdb" if ipdb or all_ else "",
        "pudb" if pudb or all_ else "",
        "icecream" if icecream or all_ else "",
        "devtools[pygments]" if devtools or all_ else "",
        "pytest-watch pytest-testmon" if watch or all_ else "",
        "pytest-watcher pytest-testmon" if watcher or all_ else "",
    ]
    run_command(c, "poetry run pip install --upgrade", *tools)

editable(c, inject='')

Hack to install a Poetry package as editable until Poetry supports PEP660 hooks.

It won't be needed anymore when https://github.com/python-poetry/poetry-core/pull/182 is merged.

Source code in src/conjuring/spells/py.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@task(help={"inject": "Pipx repo to inject this project into"})
def editable(c: Context, inject: str = "") -> None:
    """Hack to install a Poetry package as editable until Poetry supports PEP660 hooks.

    It won't be needed anymore when https://github.com/python-poetry/poetry-core/pull/182 is merged.
    """
    if not Poetry(c).used_in_project():
        return

    chosen_repo = ""
    if inject:
        # Ask for the repo before doing anything else... to fail fast if no repo is chosen
        chosen_repo = run_with_fzf(c, "ls -1 ~/.local/pipx/venvs/", query=inject)
        if not chosen_repo:
            return

    c.run("poetry build")
    c.run("tar -xvzf dist/*.gz --strip-components 1 */setup.py")
    # Ignore errors, it might not be installed
    c.run("black setup.py", warn=True)

    if not chosen_repo:
        print_error("Use --inject to inject this repo into a pipx virtualenv.")
        return

    c.run(f"mv {PYPROJECT_TOML} _{PYPROJECT_TOML}")
    run_command(c, "pipx inject -e", chosen_repo, ".")
    c.run(f"mv _{PYPROJECT_TOML} {PYPROJECT_TOML}")
    c.run("rm setup.py")

install(c, version='', force=False, delete_all=False, pipx=False, editable=False)

Install a Python virtual environment. For now, only works with Poetry.

Source code in src/conjuring/spells/py.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@task(
    help={
        "version": "Python version",
        "force": "Recreate the environment",
        "delete_all": "Delete all environments",
        "pipx": "Install with pipx",
        "editable": "Install as editable",
    },
)
def install(  # noqa: PLR0913
    c: Context,
    version: str = "",
    force: bool = False,
    delete_all: bool = False,
    pipx: bool = False,
    editable: bool = False,
) -> None:
    """Install a Python virtual environment. For now, only works with Poetry."""
    venv_list = run_lines(c, "poetry env list", hide=False)
    poetry = Poetry(c)
    if not poetry.used_in_project():
        return

    if delete_all:
        for venv_name in venv_list:
            if ".venv" in venv_name:
                c.run("rm -rf .venv")
            else:
                poetry.remove_venv(poetry.parse_python_version(venv_name))

    if not version:
        version = poetry.guess_python_version()
    pyenv = PyEnv(c)
    if force or not pyenv.has_local():
        # TODO: if tox.ini is present in the repo, set all versions from there
        pyenv.set_local(version)
    if force and not delete_all:
        poetry.remove_venv(version)
    poetry.use_venv(version)

    c.run("poetry lock --check && poetry install")
    if pipx:
        run_command(c, "pipx install", "--python", f"python{version}", " --editable" if editable else "", ".")
    c.run("poetry env list")

test(c, s=False)

Run tests with pytest.

Source code in src/conjuring/spells/py.py 
189
190
191
192
193
194
195
196
197
198
199
@task(
    help={
        "s": "Don't capture output (same shortcut as pytest)",
    },
)
def test(c: Context, s: bool = False) -> None:  # noqa: PT028
    """Run tests with pytest."""
    if not Poetry(c).used_in_project():
        return

    run_command(c, "poetry run", Pytest.command(s))

watch(c)

Watch changed files and run tests with pytest.

Source code in src/conjuring/spells/py.py 
202
203
204
205
206
207
208
@task
def watch(c: Context) -> None:
    """Watch changed files and run tests with pytest."""
    if not Poetry(c).used_in_project():
        return

    run_command(c, "poetry run", "ptw . --testmon")

conjuring.spells.shell

Shell: install/uninstall completion.

completion_install(c, app)

Install shell completion. For now, only for the Bash shell, and only for Click projects.

Source code in src/conjuring/spells/shell.py 
19
20
21
22
23
24
25
26
27
28
29
@task
def completion_install(c: Context, app: str) -> None:
    """Install shell completion. For now, only for the Bash shell, and only for Click projects.

    - [Shell Completion — Click Documentation (8.0.x)](https://click.palletsprojects.com/en/8.0.x/shell-completion/)
    - [click-contrib/click-completion: Add or enhance bash, fish, zsh and powershell completion in Click](https://github.com/click-contrib/click-completion)
    """
    completion_file = f"{COMPAT_DIR}{app}.bash-completion"
    c.run(f"_{app.upper()}_COMPLETE=bash_source {app} > {completion_file}")
    c.run(f"eza -l {completion_file}")
    c.run(f"bat {completion_file}")

completion_list(c)

List existing shell completions.

Source code in src/conjuring/spells/shell.py 
12
13
14
15
16
@task
def completion_list(c: Context) -> None:
    """List existing shell completions."""
    for var in COMPLETION_DIRS:
        c.run(f"eza -l {var}")

completion_uninstall(c, app)

Uninstall shell completion from both completion dirs.

Source code in src/conjuring/spells/shell.py 
32
33
34
35
36
37
38
@task
def completion_uninstall(c: Context, app: str) -> None:
    """Uninstall shell completion from both completion dirs."""
    for completion_dir in COMPLETION_DIRS:
        with c.cd(completion_dir):
            c.run(f"rm -v {app}*", warn=True)
    completion_list(c)