feat: add new functions and classes from dotfiles

Author: andreoliwa

clit/config.py

@@ -0,0 +1,31 @@
+"""Configuration helpers."""
+import json
+
+from clit.constants import CONFIG_DIR
+
+
+class JsonConfig:
+    """A JSON config file."""
+
+    def __init__(self, partial_path):
+        """Create or get a JSON config file inside the config directory."""
+        self.full_path = CONFIG_DIR / partial_path
+        self.full_path.parent.mkdir(parents=True, exist_ok=True)
+
+    def _generic_load(self, default):
+        """Try to load file data, and use a default when there is no data."""
+        try:
+            data = json.loads(self.full_path.read_text())
+        except (json.decoder.JSONDecodeError, FileNotFoundError):
+            data = default
+        return data
+
+    def load_set(self):
+        """Load file data as a set."""
+        return set(self._generic_load(set()))
+
+    def dump(self, new_data):
+        """Dump new JSON data in the config file."""
+        if isinstance(new_data, set):
+            new_data = list(new_data)
+        self.full_path.write_text(json.dumps(new_data))

clit/constants.py

@@ -0,0 +1,7 @@
+"""Constants."""
+from pathlib import Path
+
+SECTION_SYMLINKS_FILES = "symlinks/files"
+SECTION_SYMLINKS_DIRS = "symlinks/dirs"
+PYCHARM_APP_FULL_PATH = "/Applications/PyCharm.app/Contents/MacOS/pycharm"
+CONFIG_DIR = Path("~/.config/dotfiles/").expanduser()

clit/db.py

@@ -0,0 +1,82 @@
+"""Database module."""
+from typing import List, Optional
+
+from clit.files import shell
+
+
+class DatabaseServer:
+    """A database server URI parser."""
+
+    uri: str
+    protocol: str
+    user: Optional[str]
+    password: Optional[str]
+    server: str
+    port: Optional[int]
+
+    def __init__(self, uri):
+        """Parser the server URI and extract needed parts."""
+        self.uri = uri
+        protocol_user_password, server_port = uri.split("@")
+        self.protocol, user_password = protocol_user_password.split("://")
+        if ":" in user_password:
+            self.user, self.password = user_password.split(":")
+        else:
+            self.user, self.password = None, None
+        if ":" in server_port:
+            self.server, self.port = server_port.split(":")
+            self.port = int(self.port)
+        else:
+            self.server, self.port = server_port, None
+
+    @property
+    def uri_without_port(self):
+        """Return the URI without the port."""
+        parts = self.uri.split(":")
+        if len(parts) != 4:
+            # Return the unmodified URI if we don't have port.
+            return self.uri
+        return ":".join(parts[:-1])
+
+
+class PostgreSQLServer(DatabaseServer):
+    """A PostgreSQL database server URI parser and more stuff."""
+
+    databases: List[str] = []
+    inside_docker = False
+    psql: str = ""
+    pg_dump: str = ""
+
+    def __init__(self, *args, **kwargs):
+        """Determine which psql executable exists on this machine."""
+        super().__init__(*args, **kwargs)
+
+        self.psql = shell("which psql", quiet=True, capture_output=True).stdout
+        if not self.psql:
+            self.psql = "psql_docker"
+            self.inside_docker = True
+
+        self.pg_dump = shell("which pg_dump", quiet=True, capture_output=True).stdout
+        if not self.pg_dump:
+            self.pg_dump = "pg_dump_docker"
+            self.inside_docker = True
+
+    @property
+    def docker_uri(self):
+        """Return a URI without port if we are inside Docker."""
+        return self.uri_without_port if self.inside_docker else self.uri
+
+    def list_databases(self) -> "PostgreSQLServer":
+        """List databases."""
+        process = shell(
+            f"{self.psql} -c 'SELECT datname FROM pg_database WHERE datistemplate = false' "
+            f"--tuples-only {self.docker_uri}",
+            quiet=True,
+            capture_output=True,
+        )
+        if process.returncode:
+            print(f"Error while listing databases.\nstdout={process.stdout}\nstderr={process.stderr}")
+            exit(10)
+
+        self.databases = sorted(db.strip() for db in process.stdout.strip().split())
+        return self

clit/dev.py

@@ -0,0 +1,70 @@
+"""Development helpers."""
+import os
+from subprocess import call, check_output
+from typing import Tuple
+
+import click
+import crayons
+from plumbum import FG, RETCODE
+
+from clit.constants import PYCHARM_APP_FULL_PATH
+
+
+@click.command()
+@click.argument("files", nargs=-1)
+def pycharm_cli(files):
+    """Invoke PyCharm on the command line.
+
+    If a file doesn't exist, call `which` to find out the real location.
+    """
+    full_paths = []
+    for possible_file in files:
+        if os.path.isfile(possible_file):
+            real_file = os.path.abspath(possible_file)
+        else:
+            real_file = check_output(["which", possible_file]).decode().strip()
+        full_paths.append(real_file)
+    command_line = [PYCHARM_APP_FULL_PATH] + full_paths
+    print(crayons.green("Calling PyCharm with {}".format(" ".join(command_line))))
+    call(command_line)
+
+
+@click.command()
+@click.option("--delete", "-d", default=False, is_flag=True, help="Delete pytest directory first")
+@click.option("--failed", "-f", default=False, is_flag=True, help="Run only failed tests")
+@click.option("--count", "-c", default=0, help="Repeat the same test several times")
+@click.option("--reruns", "-r", default=0, help="Re-run a failed test several times")
+@click.argument("class_names_or_args", nargs=-1)
+def pytest_run(delete: bool, failed: bool, count: int, reruns: int, class_names_or_args: Tuple[str]):
+    """Run pytest with some shortcut options."""
+    # Import locally, so we get an error only in this function, and not in other functions of this module.
+    from plumbum.cmd import time as time_cmd, rm
+
+    if delete:
+        print(crayons.green("Removing .pytest directory", bold=True))
+        rm["-rf", ".pytest"] & FG
+
+    pytest_plus_args = ["pytest", "-vv", "--run-intermittent"]
+    if reruns:
+        pytest_plus_args.extend(["--reruns", str(reruns)])
+    if failed:
+        pytest_plus_args.append("--failed")
+
+    if count:
+        pytest_plus_args.extend(["--count", str(count)])
+
+    if class_names_or_args:
+        targets = []
+        for name in class_names_or_args:
+            if "." in name:
+                parts = name.split(".")
+                targets.append("{}.py::{}".format("/".join(parts[0:-1]), parts[-1]))
+            else:
+                # It might be an extra argument, let's just append it
+                targets.append(name)
+        pytest_plus_args.append("-s")
+        pytest_plus_args.extend(targets)
+
+    print(crayons.green("Running tests: time {}".format(" ".join(pytest_plus_args)), bold=True))
+    rv = time_cmd[pytest_plus_args] & RETCODE(FG=True)
+    exit(rv)

clit/docker.py

@@ -0,0 +1,33 @@
+"""Docker module."""
+import json
+from pathlib import Path
+from typing import List
+
+from clit.files import shell
+from clit.types import JsonDict
+
+
+class DockerContainer:
+    """A helper for Docker containers."""
+
+    def __init__(self, container_name: str) -> None:
+        """Init instance."""
+        self.container_name = container_name
+        self.inspect_json: List[JsonDict] = []
+
+    def inspect(self) -> "DockerContainer":
+        """Inspect a Docker container and save its JSON info."""
+        if not self.inspect_json:
+            raw_info = shell(f"docker inspect {self.container_name}", quiet=True, capture_output=True).stdout
+            self.inspect_json = json.loads(raw_info)
+        return self
+
+    def replace_mount_dir(self, path: Path) -> Path:
+        """Replace a mounted dir on a file/dir path inside a Docker container."""
+        self.inspect()
+        for mount in self.inspect_json[0]["Mounts"]:
+            source = mount["Source"]
+            if str(path).startswith(source):
+                new_path = str(path).replace(source, mount["Destination"])
+                return Path(new_path)
+        return path

clit/files.py

@@ -1,19 +1,19 @@
 # -*- coding: utf-8 -*-
 """Files, symbolic links, operating system utilities."""
 import os
+from argparse import ArgumentTypeError
+from pathlib import Path
 from shlex import split
-from subprocess import call, check_output
-from typing import List, Tuple
+from subprocess import PIPE, run
+from time import sleep
+from typing import List
 
 import click
 import crayons
-from plumbum import FG, RETCODE
+from plumbum import FG
 
 from clit import CONFIG, LOGGER, read_config, save_config
-
-SECTION_SYMLINKS_FILES = "symlinks/files"
-SECTION_SYMLINKS_DIRS = "symlinks/dirs"
-PYCHARM_APP_FULL_PATH = "/Applications/PyCharm.app/Contents/MacOS/pycharm"
+from clit.constants import SECTION_SYMLINKS_DIRS, SECTION_SYMLINKS_FILES
 
 
 @click.command()
@@ -109,25 +109,6 @@ def message(text, logger_func=LOGGER.warning):
     message("link created", LOGGER.info)
 
 
-@click.command()
-@click.argument("files", nargs=-1)
-def pycharm_cli(files):
-    """Invoke PyCharm on the command line.
-
-    If a file doesn't exist, call `which` to find out the real location.
-    """
-    full_paths = []
-    for possible_file in files:
-        if os.path.isfile(possible_file):
-            real_file = os.path.abspath(possible_file)
-        else:
-            real_file = check_output(["which", possible_file]).decode().strip()
-        full_paths.append(real_file)
-    command_line = [PYCHARM_APP_FULL_PATH] + full_paths
-    print(crayons.green("Calling PyCharm with {}".format(" ".join(command_line))))
-    call(command_line)
-
-
 def sync_dir(source_dirs: List[str], destination_dirs: List[str], dry_run: bool = False, kill: bool = False):
     """Synchronize a source directory with a destination."""
     # Import locally, so we get an error only in this function, and not in other functions of this module.
@@ -168,42 +149,57 @@ def backup_full(ctx, dry_run: bool, kill: bool, pictures: bool):
         print(ctx.get_help())
 
 
-@click.command()
-@click.option("--delete", "-d", default=False, is_flag=True, help="Delete pytest directory first")
-@click.option("--failed", "-f", default=False, is_flag=True, help="Run only failed tests")
-@click.option("--count", "-c", default=0, help="Repeat the same test several times")
-@click.option("--reruns", "-r", default=0, help="Re-run a failed test several times")
-@click.argument("class_names_or_args", nargs=-1)
-def pytest_run(delete: bool, failed: bool, count: int, reruns: int, class_names_or_args: Tuple[str]):
-    """Run pytest with some shortcut options."""
-    # Import locally, so we get an error only in this function, and not in other functions of this module.
-    from plumbum.cmd import time as time_cmd, rm
-
-    if delete:
-        print(crayons.green("Removing .pytest directory", bold=True))
-        rm["-rf", ".pytest"] & FG
-
-    pytest_plus_args = ["pytest", "-vv", "--run-intermittent"]
-    if reruns:
-        pytest_plus_args.extend(["--reruns", str(reruns)])
-    if failed:
-        pytest_plus_args.append("--failed")
-
-    if count:
-        pytest_plus_args.extend(["--count", str(count)])
-
-    if class_names_or_args:
-        targets = []
-        for name in class_names_or_args:
-            if "." in name:
-                parts = name.split(".")
-                targets.append("{}.py::{}".format("/".join(parts[0:-1]), parts[-1]))
-            else:
-                # It might be an extra argument, let's just append it
-                targets.append(name)
-        pytest_plus_args.append("-s")
-        pytest_plus_args.extend(targets)
-
-    print(crayons.green("Running tests: time {}".format(" ".join(pytest_plus_args)), bold=True))
-    rv = time_cmd[pytest_plus_args] & RETCODE(FG=True)
-    exit(rv)
+def shell(command_line, quiet=False, return_lines=False, **kwargs):
+    """Print and run a shell command."""
+    if not quiet:
+        print("$ {}".format(command_line))
+    if return_lines:
+        kwargs.setdefault("stdout", PIPE)
+
+    completed_process = run(command_line, shell=True, universal_newlines=True, **kwargs)
+    if not return_lines:
+        return completed_process
+
+    stdout = completed_process.stdout.strip().strip("\n")
+    return stdout.split("\n") if stdout else []
+
+
+def shell_find(command_line, **kwargs):
+    """Run a find command using the shell, and return its output as a list."""
+    if not command_line.startswith("find"):
+        command_line = f"find {command_line}"
+    kwargs.setdefault("quiet", True)
+    kwargs.setdefault("check", True)
+    return shell(command_line, return_lines=True, **kwargs)
+
+
+def _check_type(full_path, method, msg):
+    """Check a path, raise an error if it's not valid."""
+    obj = Path(full_path)
+    if not method(obj):
+        raise ArgumentTypeError(f"{full_path} is not a valid existing {msg}")
+    return obj
+
+
+def existing_directory_type(directory):
+    """Convert the string to a Path object, raising an error if it's not a directory. Use with argparse."""
+    return _check_type(directory, Path.is_dir, "directory")
+
+
+def existing_file_type(file):
+    """Convert the string to a Path object, raising an error if it's not a file. Use with argparse."""
+    return _check_type(file, Path.is_file, "file")
+
+
+def wait_for_process(process_name: str) -> None:
+    """Wait for a process to finish.
+
+    https://stackoverflow.com/questions/1058047/wait-for-any-process-to-finish
+    """
+    pid = shell(f"pidof {process_name}", quiet=True, stdout=PIPE).stdout.strip()
+    if not pid:
+        return
+
+    pid_path = Path(f"/proc/{pid}")
+    while pid_path.exists():
+        sleep(0.5)

clit/types.py

@@ -0,0 +1,4 @@
+"""Types."""
+from typing import Any, Dict
+
+JsonDict = Dict[str, Any]