Merge pull request #57 from kiviktnm/feat/symlink-parent-folders

Add option to configure symlink initial parent folder permissions

docs/README.md

@@ -127,7 +127,7 @@ The architecture of the computer's CPU. Currently, this is only used by the AUR
 decman.config.arch = "x86_64"
 ```
 
-## Files and directories
+## Files, directories and symlink
 
 Decman functions as a dotfile manager. It will install the defined files, directories and symlinks to their destinations. You can set file permissions, owners as well as define variables that will be substituted in the installed files. Decman keeps track of all files it creates and when a file is no longer present in your source, it will be also removed from its destination. This helps with keeping your system clean. However, decman won't remove directories as they might contain files that weren't created by decman.
 
@@ -208,7 +208,7 @@ Ownership, permissions, and parent directories are enforced on creation. Missing
 
 ### Symlink
 
-Declare a link to a target. Missing directories are created.
+Declare a link to a target. Missing directories are created. If you need to configure parent folder permissions, use `Symlink` objects.
 
 ```py
 import decman
@@ -216,6 +216,9 @@ import decman
 # Replaces sudo with doas
 # /usr/bin/sudo -> /usr/bin/doas
 decman.symlinks["/usr/bin/sudo"] = "/usr/bin/doas"
+
+# I don't know why would you ever do this but as an example
+decman.symlinks["/home/me/.bin/mydoas"] = decman.Symlink("/usr/bin/doas", owner="me", group="users")
 ```
 
 ## Modules
@@ -351,9 +354,10 @@ def file_variables(self) -> dict[str, str]:
 Defines symlinks fro the module.
 
 ```py
-def symlinks(self) -> dict[str, str]:
+def symlinks(self) -> dict[str, str | Symlink]:
     return {
         "/etc/resolv.conf": "/run/systemd/resolve/resolv.conf",
+        "/home/me/.config/app/file.conf": Symlink("/home/me/.file.conf", owner="me"),
     }
 ```
 

src/decman/__init__.py

@@ -3,7 +3,7 @@ import typing
 # Re-exports
 from decman.core.command import prg
 from decman.core.error import SourceError
-from decman.core.fs import Directory, File
+from decman.core.fs import Directory, File, Symlink
 from decman.core.module import Module
 from decman.core.store import Store
 from decman.plugins import Plugin, available_plugins
@@ -51,6 +51,7 @@ __all__ = [
     "SourceError",
     "File",
     "Directory",
+    "Symlink",
     "Module",
     "Store",
     "Plugin",
@@ -63,7 +64,7 @@ __all__ = [
 # -----------------------------------------
 files: dict[str, File] = {}
 directories: dict[str, Directory] = {}
-symlinks: dict[str, str] = {}
+symlinks: dict[str, str | Symlink] = {}
 modules: list[Module] = []
 execution_order: list[str] = [
     "files",

src/decman/core/file_manager.py

@@ -13,7 +13,7 @@ def update_files(
     modules: list[module.Module],
     files: dict[str, fs.File],
     directories: dict[str, fs.Directory],
-    symlinks: dict[str, str],
+    symlinks: dict[str, str | fs.Symlink],
     dry_run: bool = False,
 ) -> bool:
     """
@@ -204,39 +204,18 @@ def _install_directories(
     return checked_files, changed_files
 
 
-def _is_symlink_to(path: str, target: str) -> bool:
-    if not os.path.islink(path):
-        return False
-    return os.readlink(path) == target
-
-
 def _install_symlinks(
-    symlinks: dict[str, str], dry_run: bool = False
+    symlinks: dict[str, str | fs.Symlink], dry_run: bool = False
 ) -> tuple[list[str], list[str]]:
     checked_files = []
     changed_files = []
 
     for link_name, target in symlinks.items():
         output.print_debug(f"Checking symlink {link_name}.")
-        try:
         checked_files.append(link_name)
 
-            if _is_symlink_to(link_name, target):
+        target_link: fs.Symlink = target if type(target) is fs.Symlink else fs.Symlink(target)  # type: ignore
-                continue
+        if target_link.link_to(link_name, dry_run):
-
             changed_files.append(link_name)
 
-            if dry_run:
-                continue
-
-            if os.path.lexists(link_name):
-                os.unlink(link_name)
-
-            os.makedirs(os.path.dirname(link_name), exist_ok=True)
-            os.symlink(target, link_name)
-        except OSError as error:
-            raise errors.FSSymlinkFailedError(
-                link_name, target, error.strerror or str(error)
-            ) from error
-
     return checked_files, changed_files

src/decman/core/fs.py

@@ -8,6 +8,20 @@ import decman.core.error as errors
 import decman.core.output as output
 
 
+def create_missing_dirs(dirct: str, uid: typing.Optional[int], gid: typing.Optional[int]):
+    if not os.path.isdir(dirct):
+        parent_dir = os.path.dirname(dirct)
+        if not os.path.isdir(parent_dir):
+            create_missing_dirs(parent_dir, uid, gid)
+
+        output.print_debug(f"Creating directory '{dirct}'.")
+        os.mkdir(dirct)
+
+        if uid is not None:
+            assert gid is not None, "If uid is set, then gid is set."
+            os.chown(dirct, uid, gid)
+
+
 class File:
     """
     Declarative file specification describing how a file should be materialized at a target path.
@@ -129,21 +143,6 @@ class File:
 
         target_directory = os.path.dirname(target)
 
-        def create_missing_dirs(dirct: str, uid: typing.Optional[int], gid: typing.Optional[int]):
-            if not os.path.isdir(dirct):
-                parent_dir = os.path.dirname(dirct)
-                if not os.path.isdir(parent_dir):
-                    create_missing_dirs(parent_dir, uid, gid)
-
-                output.print_debug(
-                    f"While installing file '{target}' creating directory '{dirct}'."
-                )
-                os.mkdir(dirct)
-
-                if uid is not None:
-                    assert gid is not None, "If uid is set, then gid is set."
-                    os.chown(dirct, uid, gid)
-
         if not dry_run:
             create_missing_dirs(target_directory, self.uid, self.gid)
 
@@ -219,6 +218,95 @@ class File:
         return True
 
 
+class Symlink:
+    """
+    Declarative specification for linking a source to a destination.
+
+    Parameters:
+        ``target``:
+            Path to an existing file to serve as the target of the symlink.
+
+        ``owner``:
+            User name to own created parent directories.
+
+        ``group``:
+            Group name to own created parent directories.
+
+    Raises:
+        ``UserNotFoundError``
+            If ``owner`` does not exist on the system.
+
+        ``GroupNotFoundError``
+            If ``group`` does not exist on the system.
+    """
+
+    def __init__(
+        self,
+        target: str,
+        owner: typing.Optional[str] = None,
+        group: typing.Optional[str] = None,
+    ):
+        self.target = target
+        self.owner = owner
+        self.group = group
+        self.uid = None
+        self.gid = None
+
+        if owner is not None:
+            self.uid, self.gid = command.get_user_info(owner)
+
+        if group is not None:
+            try:
+                self.gid = grp.getgrnam(group).gr_gid
+            except KeyError as error:
+                raise errors.GroupNotFoundError(group) from error
+
+    def link_to(self, link_name: str, dry_run: bool = False) -> bool:
+        """
+        Creates a symlink ``link_name`` -> ``target``.
+
+        Parameters:
+            ``link_name``:
+                Path to the target file on disk.
+
+        Returns:
+            True if a new link was/would be created or modified.
+            False if the existing link already contained the desired target.
+
+        Raises:
+            FSSymlinkFailedError
+                If creating the symlink failed due to directory creation, file I/O, permission
+                changes, or ownership changes fail (e.g. permission denied, missing parent path
+                components, I/O errors).
+        """
+
+        def _is_symlink_to(path: str, target: str) -> bool:
+            if not os.path.islink(path):
+                return False
+            return os.readlink(path) == target
+
+        output.print_debug(f"Checking symlink {link_name}.")
+        try:
+            if _is_symlink_to(link_name, self.target):
+                return False
+
+            if dry_run:
+                return True
+
+            target_directory = os.path.dirname(link_name)
+            create_missing_dirs(target_directory, self.uid, self.gid)
+
+            if os.path.lexists(link_name):
+                os.unlink(link_name)
+
+            os.symlink(self.target, link_name)
+            return True
+        except OSError as error:
+            raise errors.FSSymlinkFailedError(
+                link_name, self.target, error.strerror or str(error)
+            ) from error
+
+
 class Directory:
     """
     Declarative specification for copying the contents of a source directory into a target

src/decman/core/module.py

@@ -105,7 +105,7 @@ class Module:
         """
         return {}
 
-    def symlinks(self) -> dict[str, str]:
+    def symlinks(self) -> dict[str, str | fs.Symlink]:
         """
         Override this method to return symlinks that should be created as a part of this
         module.