Cleanup

entangled/code_reader.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass, field
-from pathlib import Path, PurePath
+from pathlib import PurePath
 
 import mawk
 import re

entangled/commands/__init__.py

@@ -3,12 +3,14 @@
 from .stitch import stitch
 from .sync import sync
 from .tangle import tangle
+from .reset import reset
 from .watch import watch
 from .brei import brei
 
 __all__ = [
     "new",
     "brei",
+    "reset",
     "status",
     "stitch",
     "sync",

entangled/commands/brei.py

@@ -1,5 +1,6 @@
 from pathlib import Path
-from typing import Awaitable, Optional
+from collections.abc import Awaitable
+from typing import Any
 import argh  # type: ignore
 import asyncio
 import textwrap
@@ -11,15 +12,15 @@
 log = logger()
 
 
-async def main(target_strs: list[str], force_run: bool, throttle: Optional[int]):
+async def main(target_strs: list[str], force_run: bool, throttle: int | None):
     if not Path(".entangled").exists():
         Path(".entangled").mkdir()
 
-    db = await resolve_tasks(config.brei, Path(".entangled/brei_history"))
+    db = await resolve_tasks(config.get.brei, Path(".entangled/brei_history"))
     if throttle:
         db.throttle = asyncio.Semaphore(throttle)
     db.force_run = force_run
-    jobs: list[Awaitable] = [db.run(Phony(t), db=db) for t in target_strs]
+    jobs: list[Awaitable[Any]] = [db.run(Phony(t), db=db) for t in target_strs]
     with db.persistent_history():
         results = await asyncio.gather(*jobs)
 
@@ -35,8 +36,7 @@ async def main(target_strs: list[str], force_run: bool, throttle: Optional[int])
 @argh.arg("targets", nargs="+", help="name of target to run")
 @argh.arg("-B", "--force-run", help="rebuild all dependencies")
 @argh.arg("-j", "--throttle", help="limit number of concurrent jobs")
-def brei(targets: list[str], *, force_run: bool = False, throttle: Optional[int] = None):
+def brei(targets: list[str], *, force_run: bool = False, throttle: int | None = None):
     """Build one of the configured targets."""
     config.read()
     asyncio.run(main(targets, force_run, throttle))
-

entangled/commands/new.py

@@ -1,10 +1,9 @@
-from typing import Optional
-
 import argh  # type: ignore
+from argh.utils import get_subparsers
 import argparse
 
 from pathlib import Path
-from brei.cli import RichHelpFormatter
+from rich_argparse import RichHelpFormatter
 from rich.console import Console
 from rich.table import Table
 
@@ -45,7 +44,7 @@ def print_help() -> None:
     """
     parser = argparse.ArgumentParser(formatter_class=RichHelpFormatter)
     argh.add_commands(parser, [new], func_kwargs={"formatter_class": RichHelpFormatter})
-    argh.utils.get_subparsers(parser).choices["new"].print_help()
+    get_subparsers(parser).choices["new"].print_help()
 
 
 @argh.arg(
@@ -96,9 +95,9 @@ def print_help() -> None:
     help="Initialize a new project at this path",
 )
 def new(
-    template: Optional[str],
-    project_path: Optional[Path], *,
-    answers_file: Optional[str] = None,
+    template: str | None,
+    project_path: Path | None, *,
+    answers_file: str | None = None,
     data: str = "",
     defaults: bool = False,
     pretend: bool = False,
@@ -135,7 +134,7 @@ def new(
             copy_this_template = template_option.url
             break
 
-    data_dict: dict = {}
+    data_dict: dict[str, str] = {}
     if data:
         try:
             for d in data.split(";"):

entangled/commands/reset.py

@@ -0,0 +1,62 @@
+"""
+The `reset` command resets the file database in `.entangled/filedb.json`.
+This database gets updated every time you tangle or stitch, but sometimes
+its contents may become invalid, for instance when switching branches.
+This command will read the markdown sources, then pretend to be tangling
+without actually writing out to source files.
+"""
+
+from ..io import TransactionMode, transaction
+from ..config import config, get_input_files
+from ..hooks import  get_hooks
+from ..document import ReferenceMap
+from ..errors.user import UserError
+
+import logging
+from pathlib import Path
+
+
+def reset():
+    """
+    Resets the database. This performs a tangle without actually writing
+    output to the files, but updating the database as if we were.
+    """
+    config.read()
+
+    # these imports depend on config being read
+    from ..markdown_reader import read_markdown_file
+    from ..tangle import tangle_ref
+
+    input_file_list = get_input_files()
+
+    refs = ReferenceMap()
+    hooks = get_hooks()
+    logging.debug("tangling with hooks: %s", [h.__module__ for h in hooks])
+    mode = TransactionMode.RESETDB
+    annotation_method = config.get.annotation
+
+    try:
+        with transaction(mode) as t:
+            for path in input_file_list:
+                logging.debug("reading `%s`", path)
+                t.update(path)
+                _, _ = read_markdown_file(path, refs=refs, hooks=hooks)
+
+            for h in hooks:
+                h.pre_tangle(refs)
+
+            for tgt in refs.targets:
+                result, deps = tangle_ref(refs, tgt, annotation_method)
+                mask = next(iter(refs.by_name(tgt))).mode
+                t.write(Path(tgt), result, list(map(Path, deps)), mask)
+
+            for h in hooks:
+                h.on_tangle(t, refs)
+
+            t.clear_orphans()
+
+        for h in hooks:
+            h.post_tangle(refs)
+
+    except UserError as e:
+        logging.error(str(e))

entangled/commands/status.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
-from typing import Iterable
-from ..status import find_watch_dirs, list_input_files, list_dependent_files
+from collections.abc import Iterable
+from ..status import list_input_files, list_dependent_files
 from ..config import config
 from pathlib import Path
 
@@ -10,20 +10,23 @@
 from rich.panel import Panel
 from rich.tree import Tree
 
+
 def tree_from_files(files: Iterable[Path]):
     tree = Tree(label=".")
     dirs = {Path("."): tree}
     for f in sorted(files):
         for p in reversed(f.parents):
             if p not in dirs:
                 dirs[p] = dirs[p.parent].add(p.name, style="repr.path")
-        dirs[f.parent].add(f.name, style="repr.filename")
+        _ = dirs[f.parent].add(f.name, style="repr.filename")
     return tree
 
+
 def files_panel(file_list: Iterable[Path], title: str) -> Panel:
     tree = tree_from_files(file_list)
     return Panel(tree, title=title, border_style="dark_cyan")
 
+
 def rich_status():
     config_table = Table()
     config_table.add_column("name")

entangled/commands/stitch.py

@@ -6,9 +6,9 @@
 
 from ..config import config
 from ..document import ReferenceMap, Content, PlainText, ReferenceId
-from ..transaction import transaction, TransactionMode
+from ..io import transaction, TransactionMode
 from ..errors.user import UserError
-from .tangle import get_input_files
+from ..config import get_input_files
 
 
 def stitch_markdown(reference_map: ReferenceMap, content: list[Content]) -> str:
@@ -52,7 +52,7 @@ def stitch(*, force: bool = False, show: bool = False):
             content[path] = c
 
         with transaction(mode) as t:
-            for path in t.db.managed:
+            for path in t.db.managed_files:
                 logging.debug("reading `%s`", path)
                 t.update(path)
                 with open(path, "r") as f:

entangled/commands/sync.py

@@ -4,7 +4,7 @@
 
 import logging
 
-from ..filedb import file_db
+from ..io import filedb, FileCache
 from ..config import config
 from .stitch import stitch, get_input_files
 from .tangle import tangle
@@ -15,29 +15,30 @@ def _stitch_then_tangle():
     tangle()
 
 
-def sync_action() -> Optional[Callable[[], None]]:
+def sync_action() -> Callable[[], None] | None:
     input_file_list = get_input_files()
+    fs = FileCache()
 
-    with file_db(readonly=True) as db:
-        changed = set(db.changed())
+    with filedb(readonly=True) as db:
+        changed = set(db.changed_files(fs))
 
         if not all(f in db for f in input_file_list):
             return tangle
 
         if not changed:
             return None
 
-        if changed.isdisjoint(db.managed):
+        if changed.isdisjoint(db.managed_files):
             logging.info("Tangling")
             return tangle
 
-        if changed.issubset(db.managed):
+        if changed.issubset(db.managed_files):
             logging.info("Stitching")
             return _stitch_then_tangle
 
         logging.error("changed: %s", [str(p) for p in changed])
         logging.error(
-            "Both markdown and code seem to have changed. " "Don't know what to do now."
+            "Both markdown and code seem to have changed, don't know what to do now."
         )
         return None
 

entangled/commands/tangle.py

@@ -1,25 +1,15 @@
-from itertools import chain
 from pathlib import Path
 
 import argh  # type: ignore
 import logging
 
 from ..document import ReferenceMap
-from ..config import config, AnnotationMethod
-from ..transaction import transaction, TransactionMode
+from ..config import config, AnnotationMethod, get_input_files
+from ..io import transaction, TransactionMode
 from ..hooks import get_hooks
 from ..errors.user import UserError
 
 
-def get_input_files() -> list[Path]:
-    include_file_list = chain.from_iterable(map(Path(".").glob, config.get.watch_list))
-    input_file_list = [
-        path for path in include_file_list
-        if not any(path.match(pat) for pat in config.get.ignore_list)
-    ]
-    return input_file_list
-
-
 @argh.arg(
     "-a",
     "--annotate",
@@ -62,7 +52,7 @@ def tangle(*, annotate: str | None = None, force: bool = False, show: bool = Fal
             for path in input_file_list:
                 logging.debug("reading `%s`", path)
                 t.update(path)
-                read_markdown_file(path, refs=refs, hooks=hooks)
+                _, _ = read_markdown_file(path, refs=refs, hooks=hooks)
 
             for h in hooks:
                 h.pre_tangle(refs)

entangled/config/__init__.py

@@ -11,6 +11,7 @@
 from enum import StrEnum
 from pathlib import Path
 from typing import Any
+from itertools import chain
 
 import msgspec
 from msgspec import Struct, field
@@ -86,7 +87,7 @@ class Config(Struct, dict=True):
     annotation: AnnotationMethod = AnnotationMethod.STANDARD
     use_line_directives: bool = False
     hooks: list[str] = field(default_factory=lambda: ["shebang"])
-    hook: dict[str, Any] = field(default_factory=dict)
+    hook: dict[str, Any] = field(default_factory=dict)  # pyright: ignore[reportExplicitAny]
     brei: Program = field(default_factory=Program)
 
     language_index: dict[str, Language] = field(default_factory=dict)
@@ -126,10 +127,10 @@ def read_config_from_toml(
         return None
     try:
         with open(path, "rb") as f:
-            json: Any = tomllib.load(f)
+            json: Any = tomllib.load(f)  # pyright: ignore[reportExplicitAny]
             if section is not None:
                 for s in section.split("."):
-                    json = json[s]
+                    json = json[s]  # pyright: ignore[reportAny]
             return msgspec.convert(json, type=Config, dec_hook=from_str.dec_hook)
 
     except ValueError as e:
@@ -162,7 +163,7 @@ def read(self, force: bool = False):
     @property
     def get(self) -> Config:
         if self.config is None:
-            raise ValueError(f"No config loaded.")
+            raise ValueError("No config loaded.")
         return self.config
 
     @contextmanager
@@ -179,10 +180,19 @@ def __call__(self, **kwargs):
 
     def get_language(self, lang_name: str) -> Language | None:
         if self.config is None:
-            raise ValueError(f"No config loaded.")
+            raise ValueError("No config loaded.")
         return self.config.language_index.get(lang_name, None)
 
 
 config = ConfigWrapper()
 """The `config.config` variable is changed when the `config` module is loaded.
 Config is read from `entangled.toml` file."""
+
+
+def get_input_files() -> list[Path]:
+    include_file_list = chain.from_iterable(map(Path(".").glob, config.get.watch_list))
+    input_file_list = [
+        path for path in include_file_list
+        if not any(path.match(pat) for pat in config.get.ignore_list)
+    ]
+    return input_file_list

entangled/document.py

@@ -125,8 +125,10 @@ def content_to_text(r: ReferenceMap, c: Content) -> str:
         A string, usually not terminated by a newline.
     """
     match c:
-        case PlainText(s): return s
-        case ReferenceId(): return r.get_codeblock(c).indented_text
+        case PlainText(s):
+            return s
+        case ReferenceId():
+            return r.get_codeblock(c).indented_text
 
 
 def document_to_text(r: ReferenceMap, cs: Iterable[Content]) -> str: