[IMP] databases: link a repository to a database from the command line

docs/tutorials/commands.md

@@ -18,6 +18,7 @@ In this basic tutorial, we'll see how we can define a brand new command to print
     -   [Going further...](#going-further)
         -   [Extended command classes](#extended-command-classes)
         -   [Mutually exclusive arguments](#mutually-exclusive-arguments)
+        -   [Flags that can be turned off](#flags-that-can-be-turned-off)
         -   [Unknown arguments](#unknown-arguments)
         -   [Raising errors](#raising-errors)
 
@@ -265,6 +266,32 @@ class SampleCommand(Command):
 
 ![odev sample](img/command-sample-08.png)
 
+### Flags that can be turned off
+
+A flag defined with `args.Flag` is either present or absent, so a command cannot tell "the user asked for the value to
+be turned off" apart from "the user did not mention it". When that distinction matters, typically for a command editing
+a value that already exists, use `args.FlagOptional` instead: it registers a `--no-` counterpart for each of its
+aliases.
+
+```python
+class SampleCommand(Command):
+    """Example command used for tutorials purposes."""
+
+    _name = "sample"
+    _aliases = ["example"]
+
+    flag = args.FlagOptional(aliases=["--flag"], description="Sample three-state flag argument")
+
+    def run(self):
+        if self.args.flag is None:
+            self.console.print("The flag was not mentioned, leaving the value as it is")
+        else:
+            self.console.print(f"The flag was set to {self.args.flag}")
+```
+
+`--flag` sets the value to `True`, `--no-flag` sets it to `False`, and omitting both leaves it at its default, `None`
+unless another one is given to the argument.
+
 ### Unknown arguments
 
 By default, Odev will treat any unknown argument received as invalid and raise an error.

odev/_version.py

@@ -22,4 +22,4 @@
 # or merged change.
 # ------------------------------------------------------------------------------
 
-__version__ = "4.29.9"
+__version__ = "4.30.0"

odev/commands/database/database.py

@@ -0,0 +1,187 @@
+"""Edit the parameters of a local database without starting it."""
+
+from odev.common import args, string
+from odev.common.commands import GitCommand, LocalDatabaseCommand
+from odev.common.connectors import GitConnector
+from odev.common.errors import ConnectorError
+from odev.common.logging import logging
+from odev.common.python import PythonEnv
+
+
+logger = logging.getLogger(__name__)
+
+
+class DatabaseSetCommand(LocalDatabaseCommand, GitCommand):
+    """Display and edit the parameters of a local database without starting it.
+    Called without any argument other than the database, the current parameters are displayed.
+    """
+
+    _name = "database"
+    _aliases = ["db"]
+
+    set_repository = args.String(
+        aliases=["--set-repo"],
+        description="""Change the repository linked to the database. Accepts a repository name in
+        the format <organization>/<repository>, a git URL or the path to a local clone.
+        """,
+        metavar="REPOSITORY",
+    )
+    remove_repository = args.Flag(
+        aliases=["--remove-repo"],
+        description="Remove the repository linked to the database.",
+    )
+
+    set_venv = args.String(
+        aliases=["--set-venv"],
+        description="Change the virtualenv linked to the database.",
+        metavar="VENV",
+    )
+    remove_venv = args.Flag(
+        aliases=["--remove-venv"],
+        description="Remove the virtualenv linked to the database.",
+    )
+
+    set_worktree = args.String(
+        aliases=["--set-worktree"],
+        description="Change the worktree linked to the database.",
+        metavar="WORKTREE",
+    )
+    remove_worktree = args.Flag(
+        aliases=["--remove-worktree"],
+        description="Remove the worktree linked to the database.",
+    )
+
+    whitelist = args.FlagOptional(
+        aliases=["--whitelist"],
+        description="Whitelist or unwhitelist the database, preventing or allowing its automatic removal.",
+    )
+
+    _exclusive_pairs = (
+        ("set_repository", "remove_repository"),
+        ("set_venv", "remove_venv"),
+        ("set_worktree", "remove_worktree"),
+    )
+    """Pairs of arguments that set and remove the same value, and cannot be used together."""
+
+    @classmethod
+    def prepare_command(cls, *args, **kwargs) -> None:
+        super().prepare_command(*args, **kwargs)
+        cls.remove_argument("version")
+
+    def run(self):
+        self._check_exclusive_arguments()
+
+        if self._has_changes:
+            self._set_values()
+            self._remove_values()
+
+        self._print_values()
+
+    @property
+    def _has_changes(self) -> bool:
+        """Whether the command was called with an argument changing a parameter."""
+        return bool(
+            self.args.set_repository
+            or self.args.remove_repository
+            or self.args.set_venv
+            or self.args.remove_venv
+            or self.args.set_worktree
+            or self.args.remove_worktree
+            or self.args.whitelist is not None
+        )
+
+    def _check_exclusive_arguments(self) -> None:
+        """Ensure no parameter is both set and removed in the same call."""
+        for set_argument, remove_argument in self._exclusive_pairs:
+            if getattr(self.args, set_argument) and getattr(self.args, remove_argument):
+                raise self.error(
+                    f"Arguments {self.argument_name(set_argument)!r} and "
+                    f"{self.argument_name(remove_argument)!r} cannot be used together"
+                )
+
+    def argument_name(self, argument: str) -> str:
+        """Return the CLI alias of an argument, for use in error messages."""
+        aliases = self._arguments.get(argument, {}).get("aliases", [])
+        return next((alias for alias in aliases if alias.startswith("--")), argument)
+
+    def _set_values(self) -> None:
+        if self.args.set_repository:
+            self._set_repository(self.args.set_repository)
+
+        if self.args.set_venv:
+            venv = PythonEnv(self.args.set_venv)
+
+            if not venv.exists:
+                raise self.error(f"Virtualenv {self.args.set_venv!r} not found, please create it and retry")
+
+            self._database.venv = venv
+            logger.info(f"Virtualenv set to {venv.name!r}")
+
+        if self.args.set_worktree:
+            if self.args.set_worktree not in self.grouped_worktrees:
+                raise self.error(f"Worktree {self.args.set_worktree!r} not found, please create it and retry")
+
+            self._database.worktree = self.args.set_worktree
+            logger.info(f"Worktree set to {self.args.set_worktree!r}")
+
+        if self.args.whitelist is True:
+            self._database.whitelisted = True
+            logger.info("Database whitelisted")
+
+    def _set_repository(self, repository: str) -> None:
+        """Link a repository to the database, cloning it first if it is missing locally.
+
+        :param repository: The repository name, git URL or path to a local clone.
+        """
+        try:
+            connector = GitConnector(repository)
+        except ConnectorError as error:
+            raise self.error(str(error)) from error
+
+        if not connector.exists and self.console.confirm(
+            f"Repository {connector.name!r} not found locally, clone now?"
+        ):
+            self.odev.run_command("clone", connector.name)
+
+        linked = self._database.link_repository(connector)
+        logger.info(f"Repository set to {linked.full_name!r}")
+
+    def _remove_values(self) -> None:
+        # Values are cleared through the data store rather than through the database properties:
+        # those fall back to reading the store when their cached value is empty, so assigning None
+        # to them would write the value that was just cleared straight back.
+        if self.args.remove_repository:
+            self.store.databases.set_value(self._database, "repository", None)
+            self.store.databases.set_value(self._database, "branch", None)
+            self._database._repository = None
+            self._database._branch = None
+            logger.info("Repository removed")
+
+        if self.args.remove_venv:
+            self.store.databases.set_value(self._database, "virtualenv", None)
+            self._database._venv = None
+            logger.info("Virtualenv removed")
+
+        if self.args.remove_worktree:
+            self.store.databases.set_value(self._database, "worktree", None)
+            self._database._worktree = None
+            logger.info("Worktree removed")
+
+        if self.args.whitelist is False:
+            self._database.whitelisted = False
+            logger.info("Database unwhitelisted")
+
+    def _print_values(self) -> None:
+        """Print the current parameters of the database."""
+        info = self.store.databases.get(self._database)
+        values = {
+            "Repository": info and info.repository,
+            "Branch": info and info.branch,
+            "Virtualenv": info and info.virtualenv,
+            "Worktree": info and info.worktree,
+            "Whitelisted": "yes" if info and info.whitelisted else "no",
+        }
+        logger.info(
+            f"Parameters of database {self._database.name!r}:\n"
+            + string.join_bullet([f"{key}: {value or 'not set'}" for key, value in values.items()])
+        )

odev/common/arguments.py

@@ -2,11 +2,9 @@
 
 import pathlib
 import re
+from argparse import Action, BooleanOptionalAction
 from collections.abc import MutableMapping
-from typing import (
-    Any,
-    Literal,
-)
+from typing import Any, Literal
 
 
 class Argument:
@@ -30,7 +28,8 @@ def __init__(  # noqa: PLR0913
             "store_path",
             "store_regex",
             "store_eval",
-        ] = "store",
+        ]
+        | type[Action] = "store",
         **kwargs: Any,
     ) -> None:
         """Initialize the argument and converts it to a mapping that can be fed to the command's
@@ -197,11 +196,56 @@ def __init__(
         :param kwargs: Additional keyword arguments to pass to the ArgumentParser.
             See: https://docs.python.org/3/library/argparse.html#quick-links-for-add-argument
         """
+        action = kwargs.pop("action", None)
+
+        if action is None:
+            # A plain flag has no value of its own: the default decides which way toggling it goes,
+            # and argparse infers the resulting default from the action.
+            action = "store_false" if default is True else "store_true"
+        else:
+            # An explicit action consumes the default itself, so it has to be forwarded.
+            kwargs["default"] = default
+
+        super().__init__(
+            name=name,
+            aliases=aliases,
+            description=description,
+            action=action,
+            **kwargs,
+        )
+
+
+class FlagOptional(Flag):
+    """Flag with a three-state boolean value, registering both `--flag` and `--no-flag`."""
+
+    def __init__(
+        self,
+        name: str | None = None,
+        aliases: list[str] | None = None,
+        description: str | None = None,
+        default: bool | None = None,
+        **kwargs: Any,
+    ) -> None:
+        """Add a flag that can be set, unset or left alone.
+
+        Unlike :class:`Flag`, which is either present or absent, this registers a `--no-` counterpart
+        for each of its aliases, so that `--flag` sets the value to True, `--no-flag` sets it to False
+        and omitting both leaves it at its default, `None` unless specified otherwise. Use it to tell
+        "the user asked for the value to be turned off" apart from "the user did not mention it".
+
+        :param name: The name of the argument, will be used in the help command and in the command's class `args` attribute.
+        :param aliases: The aliases for the argument.
+        :param description: A description for the argument, will be displayed in the `help` command.
+        :param default: The value the argument takes when neither the flag nor its counterpart is present.
+        :param kwargs: Additional keyword arguments to pass to the ArgumentParser.
+            See: https://docs.python.org/3/library/argparse.html#quick-links-for-add-argument
+        """
         super().__init__(
             name=name,
             aliases=aliases,
             description=description,
-            action="store_false" if default is True else "store_true",
+            action=BooleanOptionalAction,
+            default=default,
             **kwargs,
         )
 

odev/common/commands/git.py

@@ -29,7 +29,7 @@ def worktrees(self) -> Generator[GitWorktree, None, None]:
                 if not worktree.path.exists():
                     logger.debug(f"Skipping missing worktree {worktree.name!r} at {worktree.path!s}")
                     continue
-                if hasattr(self, "args") and self.args.version and worktree.name != self.args.version:
+                if getattr(getattr(self, "args", None), "version", None) and worktree.name != self.args.version:
                     continue
                 yield worktree
 

odev/common/databases/local.py

@@ -25,7 +25,7 @@
 from packaging.version import Version
 
 from odev.common import bash, progress, string
-from odev.common.connectors import GitWorktree, PostgresConnector
+from odev.common.connectors import GitConnector, GitWorktree, PostgresConnector
 from odev.common.databases import Branch, Database, Filestore, Repository
 from odev.common.databases.base import DatabaseInfoSection
 from odev.common.errors import OdevError
@@ -362,11 +362,36 @@ def repository(self) -> Repository | None:
         return self._repository
 
     @repository.setter
-    def repository(self, value: Repository):
+    def repository(self, value: Repository | None):
         """Set the repository of the database."""
         self._repository = value
+        # The branch is a branch *of the repository*, keeping a cached one would persist
+        # the branch of the previous repository under the new one.
+        self._branch = None
         self.store.databases.set(self)
 
+    def link_repository(self, repository: str | Repository | GitConnector | None) -> Repository | None:
+        """Link a git repository to this database and save the link in the data store.
+
+        The value is normalized through :class:`GitConnector`, so repository names, HTTPS and SSH
+        URLs and paths to local clones are all accepted and stored as `organization/repository`.
+
+        :param repository: The repository to link, or None to unlink the current one.
+        :return: The linked repository, or None if it was unlinked.
+        :rtype: Optional[Repository]
+        """
+        if repository is None:
+            self.repository = None
+            return None
+
+        if not isinstance(repository, GitConnector):
+            repository = GitConnector(
+                repository.full_name if isinstance(repository, Repository) else repository,
+            )
+
+        self.repository = Repository(name=repository._repository, organization=repository._organization)
+        return self._repository
+
     @property
     def branch(self) -> Branch | None:
         if self.repository is None:

odev/common/postgres.py

@@ -1,6 +1,6 @@
 """PostgreSQL database class."""
 
-from collections.abc import Mapping, MutableMapping
+from collections.abc import Mapping, MutableMapping, Sequence
 from contextlib import nullcontext
 
 from psycopg2.errors import InvalidTableDefinition
@@ -111,10 +111,15 @@ def create_column(self, table: str, column: str, definition: str):
         return self.connector.create_column(table, column, definition)
 
     @ensure_connected
-    def query(self, query: str, nocache: bool = False):
-        """Execute a query on the database."""
+    def query(self, query: str, params: Sequence | None = None, nocache: bool = False):
+        """Execute a query on the database.
+
+        :param query: The query to execute.
+        :param params: Values to bind to the placeholders of the query.
+        :param nocache: Whether to bypass the query cache.
+        """
         with self.connector.nocache() if nocache else nullcontext():
-            return self.connector.query(query)
+            return self.connector.query(query, params)
 
     @ensure_connected
     def constraint(self, table: str, name: str, definition: str):