API Reference

Note

There's no stability guarantee as this is just for internal purposes currently.

Core Modules

rustfava.core

This module provides the data required by Fava's reports.

EntryNotFoundForHashError

Bases: RustfavaAPIError

Entry not found for hash.

Source code in src/rustfava/core/__init__.py

class EntryNotFoundForHashError(RustfavaAPIError):
    """Entry not found for hash."""

    def __init__(self, entry_hash: str) -> None:
        super().__init__(f'No entry found for hash "{entry_hash}"')

StatementNotFoundError

Bases: RustfavaAPIError

Statement not found.

Source code in src/rustfava/core/__init__.py

class StatementNotFoundError(RustfavaAPIError):
    """Statement not found."""

    def __init__(self) -> None:
        super().__init__("Statement not found.")

StatementMetadataInvalidError

Bases: RustfavaAPIError

Statement metadata not found or invalid.

Source code in src/rustfava/core/__init__.py

class StatementMetadataInvalidError(RustfavaAPIError):
    """Statement metadata not found or invalid."""

    def __init__(self, key: str) -> None:
        super().__init__(
            f"Statement path at key '{key}' missing or not a string."
        )

JournalPage dataclass

A page of journal entries.

Source code in src/rustfava/core/__init__.py

@dataclass(frozen=True)
class JournalPage:
    """A page of journal entries."""

    entries: Sequence[tuple[int, Directive]]
    total_pages: int

FilteredLedger

Filtered Beancount ledger.

Source code in src/rustfava/core/__init__.py

class FilteredLedger:
    """Filtered Beancount ledger."""

    __slots__ = (
        "__dict__",  # for the cached_property decorator
        "_date_first",
        "_date_last",
        "_pages",
        "date_range",
        "entries",
        "ledger",
    )
    _date_first: date | None
    _date_last: date | None

    def __init__(
        self,
        ledger: RustfavaLedger,
        *,
        account: str | None = None,
        filter: str | None = None,  # noqa: A002
        time: str | None = None,
    ) -> None:
        """Create a filtered view of a ledger.

        Args:
            ledger: The ledger to filter.
            account: The account filter.
            filter: The advanced filter.
            time: The time filter.
        """
        self.ledger = ledger
        self.date_range: DateRange | None = None
        self._pages: (
            tuple[
                int,
                Literal["asc", "desc"],
                list[Sequence[tuple[int, Directive]]],
            ]
            | None
        ) = None

        entries = ledger.all_entries
        if account:
            entries = AccountFilter(account).apply(entries)
        if filter and filter.strip():
            entries = AdvancedFilter(filter.strip()).apply(entries)
        if time:
            time_filter = TimeFilter(ledger.options, ledger.fava_options, time)
            entries = time_filter.apply(entries)
            self.date_range = time_filter.date_range
        self.entries = entries

        if self.date_range:
            self._date_first = self.date_range.begin
            self._date_last = self.date_range.end
            return

        self._date_first = None
        self._date_last = None
        for entry in self.entries:
            if isinstance(entry, Transaction):
                self._date_first = entry.date
                break
        for entry in reversed(self.entries):
            if isinstance(entry, (Transaction, Price)):
                self._date_last = entry.date + timedelta(1)
                break

    @property
    def end_date(self) -> date | None:
        """The date to use for prices."""
        date_range = self.date_range
        if date_range:
            return date_range.end_inclusive
        return None

    @cached_property
    def entries_with_all_prices(self) -> Sequence[Directive]:
        """The filtered entries, with all prices added back in for queries."""
        entries = [*self.entries, *self.ledger.all_entries_by_type.Price]
        entries.sort(key=_incomplete_sortkey)
        return entries

    @cached_property
    def entries_without_prices(self) -> Sequence[Directive]:
        """The filtered entries, without prices for journals."""
        return [e for e in self.entries if not isinstance(e, Price)]

    @cached_property
    def root_tree(self) -> Tree:
        """A root tree."""
        return Tree(self.entries)

    @cached_property
    def root_tree_closed(self) -> Tree:
        """A root tree for the balance sheet."""
        tree = Tree(self.entries)
        tree.cap(self.ledger.options, self.ledger.fava_options.unrealized)
        return tree

    def interval_ranges(self, interval: Interval) -> Sequence[DateRange]:
        """Yield date ranges corresponding to interval boundaries.

        Args:
            interval: The interval to yield ranges for.
        """
        if not self._date_first or not self._date_last:
            return []
        complete = not self.date_range
        return dateranges(
            self._date_first, self._date_last, interval, complete=complete
        )

    def prices(self, base: str, quote: str) -> Sequence[tuple[date, Decimal]]:
        """List all prices for a pair of commodities.

        Args:
            base: The price base.
            quote: The price quote.
        """
        all_prices = self.ledger.prices.get_all_prices((base, quote))
        if all_prices is None:
            return []

        date_range = self.date_range
        if date_range:
            return [
                price_point
                for price_point in all_prices
                if date_range.begin <= price_point[0] < date_range.end
            ]
        return all_prices

    def account_is_closed(self, account_name: str) -> bool:
        """Check if the account is closed.

        Args:
            account_name: An account name.

        Returns:
            True if the account is closed before the end date of the current
            time filter.
        """
        date_range = self.date_range
        close_date = self.ledger.accounts[account_name].close_date
        if close_date is None:
            return False
        return close_date < date_range.end if date_range else True

    def paginate_journal(
        self,
        page: int,
        per_page: int = 1000,
        order: Literal["asc", "desc"] = "desc",
    ) -> JournalPage | None:
        """Get entries for a journal page with pagination info.

        Args:
            page: Page number (1-indexed).
            order: Datewise order to sort in
            per_page: Number of entries per page.

        Returns:
            A JournalPage, containing a list of entries as (global_index,
            directive) tuples in reverse chronological order and the total
            number of pages.
        """
        if (
            self._pages is None
            or self._pages[0] != per_page
            or self._pages[1] != order
        ):
            pages: list[Sequence[tuple[int, Directive]]] = []
            enumerated = list(enumerate(self.entries_without_prices))
            entries = (
                iter(enumerated) if order == "asc" else reversed(enumerated)
            )
            while batch := tuple(islice(entries, per_page)):
                pages.append(batch)
            if not pages:
                pages.append([])
            self._pages = (per_page, order, pages)
        _per_pages, _order, pages = self._pages
        total = len(pages)
        if page > total:
            return None
        return JournalPage(pages[page - 1], total)

end_date property

The date to use for prices.

entries_with_all_prices cached property

The filtered entries, with all prices added back in for queries.

entries_without_prices cached property

The filtered entries, without prices for journals.

root_tree cached property

A root tree.

root_tree_closed cached property

A root tree for the balance sheet.

__init__(ledger, *, account=None, filter=None, time=None)

Create a filtered view of a ledger.

Parameters:

Name	Type	Description	Default
ledger	RustfavaLedger	The ledger to filter.	required
account	str | None	The account filter.	None
filter	str | None	The advanced filter.	None
time	str | None	The time filter.	None

Source code in src/rustfava/core/__init__.py

def __init__(
    self,
    ledger: RustfavaLedger,
    *,
    account: str | None = None,
    filter: str | None = None,  # noqa: A002
    time: str | None = None,
) -> None:
    """Create a filtered view of a ledger.

    Args:
        ledger: The ledger to filter.
        account: The account filter.
        filter: The advanced filter.
        time: The time filter.
    """
    self.ledger = ledger
    self.date_range: DateRange | None = None
    self._pages: (
        tuple[
            int,
            Literal["asc", "desc"],
            list[Sequence[tuple[int, Directive]]],
        ]
        | None
    ) = None

    entries = ledger.all_entries
    if account:
        entries = AccountFilter(account).apply(entries)
    if filter and filter.strip():
        entries = AdvancedFilter(filter.strip()).apply(entries)
    if time:
        time_filter = TimeFilter(ledger.options, ledger.fava_options, time)
        entries = time_filter.apply(entries)
        self.date_range = time_filter.date_range
    self.entries = entries

    if self.date_range:
        self._date_first = self.date_range.begin
        self._date_last = self.date_range.end
        return

    self._date_first = None
    self._date_last = None
    for entry in self.entries:
        if isinstance(entry, Transaction):
            self._date_first = entry.date
            break
    for entry in reversed(self.entries):
        if isinstance(entry, (Transaction, Price)):
            self._date_last = entry.date + timedelta(1)
            break

interval_ranges(interval)

Yield date ranges corresponding to interval boundaries.

Parameters:

Name	Type	Description	Default
interval	Interval	The interval to yield ranges for.	required

Source code in src/rustfava/core/__init__.py

def interval_ranges(self, interval: Interval) -> Sequence[DateRange]:
    """Yield date ranges corresponding to interval boundaries.

    Args:
        interval: The interval to yield ranges for.
    """
    if not self._date_first or not self._date_last:
        return []
    complete = not self.date_range
    return dateranges(
        self._date_first, self._date_last, interval, complete=complete
    )

prices(base, quote)

List all prices for a pair of commodities.

Parameters:

Name	Type	Description	Default
base	str	The price base.	required
quote	str	The price quote.	required

Source code in src/rustfava/core/__init__.py

def prices(self, base: str, quote: str) -> Sequence[tuple[date, Decimal]]:
    """List all prices for a pair of commodities.

    Args:
        base: The price base.
        quote: The price quote.
    """
    all_prices = self.ledger.prices.get_all_prices((base, quote))
    if all_prices is None:
        return []

    date_range = self.date_range
    if date_range:
        return [
            price_point
            for price_point in all_prices
            if date_range.begin <= price_point[0] < date_range.end
        ]
    return all_prices

account_is_closed(account_name)

Check if the account is closed.

Parameters:

Name	Type	Description	Default
account_name	str	An account name.	required

Returns:

Type	Description
bool	True if the account is closed before the end date of the current
bool	time filter.

Source code in src/rustfava/core/__init__.py

def account_is_closed(self, account_name: str) -> bool:
    """Check if the account is closed.

    Args:
        account_name: An account name.

    Returns:
        True if the account is closed before the end date of the current
        time filter.
    """
    date_range = self.date_range
    close_date = self.ledger.accounts[account_name].close_date
    if close_date is None:
        return False
    return close_date < date_range.end if date_range else True

paginate_journal(page, per_page=1000, order='desc')

Get entries for a journal page with pagination info.

Parameters:

Name	Type	Description	Default
page	int	Page number (1-indexed).	required
order	Literal['asc', 'desc']	Datewise order to sort in	'desc'
per_page	int	Number of entries per page.	1000

Returns:

Type	Description
JournalPage | None	A JournalPage, containing a list of entries as (global_index,
JournalPage | None	directive) tuples in reverse chronological order and the total
JournalPage | None	number of pages.

Source code in src/rustfava/core/__init__.py

def paginate_journal(
    self,
    page: int,
    per_page: int = 1000,
    order: Literal["asc", "desc"] = "desc",
) -> JournalPage | None:
    """Get entries for a journal page with pagination info.

    Args:
        page: Page number (1-indexed).
        order: Datewise order to sort in
        per_page: Number of entries per page.

    Returns:
        A JournalPage, containing a list of entries as (global_index,
        directive) tuples in reverse chronological order and the total
        number of pages.
    """
    if (
        self._pages is None
        or self._pages[0] != per_page
        or self._pages[1] != order
    ):
        pages: list[Sequence[tuple[int, Directive]]] = []
        enumerated = list(enumerate(self.entries_without_prices))
        entries = (
            iter(enumerated) if order == "asc" else reversed(enumerated)
        )
        while batch := tuple(islice(entries, per_page)):
            pages.append(batch)
        if not pages:
            pages.append([])
        self._pages = (per_page, order, pages)
    _per_pages, _order, pages = self._pages
    total = len(pages)
    if page > total:
        return None
    return JournalPage(pages[page - 1], total)

RustfavaLedger

Interface for a Beancount ledger.

Source code in src/rustfava/core/__init__.py

class RustfavaLedger:
    """Interface for a Beancount ledger."""

    __slots__ = (
        "_is_encrypted",
        "accounts",
        "accounts",
        "all_entries",
        "all_entries_by_type",
        "attributes",
        "beancount_file_path",
        "budgets",
        "charts",
        "commodities",
        "extensions",
        "fava_options",
        "fava_options_errors",
        "file",
        "format_decimal",
        "get_entry",
        "get_filtered",
        "ingest",
        "load_errors",
        "misc",
        "options",
        "prices",
        "query_shell",
        "watcher",
    )

    #: List of all (unfiltered) entries.
    all_entries: Sequence[Directive]

    #: A list of all errors reported by Beancount.
    load_errors: Sequence[BeancountError]

    #: The Beancount options map.
    options: BeancountOptions

    #: A dict with all of Fava's option values.
    fava_options: RustfavaOptions

    #: A list of all errors from parsing the custom options.
    fava_options_errors: Sequence[BeancountError]

    #: The price map.
    prices: RustfavaPriceMap

    #: Dict of list of all (unfiltered) entries by type.
    all_entries_by_type: EntriesByType

    #: A :class:`.AccountDict` module - details about the accounts.
    accounts: AccountDict

    #: An :class:`AttributesModule` instance.
    attributes: AttributesModule

    #: A :class:`.BudgetModule` instance.
    budgets: BudgetModule

    #: A :class:`.ChartModule` instance.
    charts: ChartModule

    #: A :class:`.CommoditiesModule` instance.
    commodities: CommoditiesModule

    #: A :class:`.ExtensionModule` instance.
    extensions: ExtensionModule

    #: A :class:`.FileModule` instance.
    file: FileModule

    #: A :class:`.DecimalFormatModule` instance.
    format_decimal: DecimalFormatModule

    #: A :class:`.IngestModule` instance.
    ingest: IngestModule

    #: A :class:`.FavaMisc` instance.
    misc: FavaMisc

    #: A :class:`.QueryShell` instance.
    query_shell: QueryShell

    def __init__(self, path: str, *, poll_watcher: bool = False) -> None:
        """Create an interface for a Beancount ledger.

        Arguments:
            path: Path to the main Beancount file.
            poll_watcher: Whether to use the polling file watcher.
        """
        #: The path to the main Beancount file.
        self.beancount_file_path = path
        self._is_encrypted = is_encrypted_file(path)
        self.get_filtered = lru_cache(maxsize=16)(self._get_filtered)
        self.get_entry = lru_cache(maxsize=16)(self._get_entry)

        self.accounts = AccountDict(self)
        self.attributes = AttributesModule(self)
        self.budgets = BudgetModule(self)
        self.charts = ChartModule(self)
        self.commodities = CommoditiesModule(self)
        self.extensions = ExtensionModule(self)
        self.file = FileModule(self)
        self.format_decimal = DecimalFormatModule(self)
        self.ingest = IngestModule(self)
        self.misc = FavaMisc(self)
        self.query_shell = QueryShell(self)

        self.watcher = WatchfilesWatcher() if not poll_watcher else Watcher()

        self.load_file()

    def load_file(self) -> None:
        """Load the main file and all included files and set attributes."""
        self.all_entries, self.load_errors, self.options = load_uncached(
            self.beancount_file_path,
            is_encrypted=self._is_encrypted,
        )
        self.get_filtered.cache_clear()
        self.get_entry.cache_clear()

        self.all_entries_by_type = group_entries_by_type(self.all_entries)
        self.prices = RustfavaPriceMap(self.all_entries_by_type.Price)

        self.fava_options, self.fava_options_errors = parse_options(
            self.all_entries_by_type.Custom,
        )

        if self._is_encrypted:  # pragma: no cover
            pass
        else:
            self.watcher.update(*self.paths_to_watch())

        # Call load_file of all modules.
        self.accounts.load_file()
        self.attributes.load_file()
        self.budgets.load_file()
        self.charts.load_file()
        self.commodities.load_file()
        self.extensions.load_file()
        self.file.load_file()
        self.format_decimal.load_file()
        self.misc.load_file()
        self.query_shell.load_file()
        self.ingest.load_file()

        self.extensions.after_load_file()

    def _get_filtered(
        self,
        account: str | None = None,
        filter: str | None = None,  # noqa: A002
        time: str | None = None,
    ) -> FilteredLedger:
        """Filter the ledger.

        Args:
            account: The account filter.
            filter: The advanced filter.
            time: The time filter.
        """
        return FilteredLedger(
            ledger=self, account=account, filter=filter, time=time
        )

    @property
    def mtime(self) -> int:
        """The timestamp to the latest change of the underlying files."""
        return self.watcher.last_checked

    @property
    def errors(self) -> Sequence[BeancountError]:
        """The errors that the Beancount loading plus Fava module errors."""
        return [
            *self.load_errors,
            *self.fava_options_errors,
            *self.budgets.errors,
            *self.extensions.errors,
            *self.misc.errors,
            *self.ingest.errors,
        ]

    @property
    def root_accounts(self) -> tuple[str, str, str, str, str]:
        """The five root accounts."""
        options = self.options
        return (
            options["name_assets"],
            options["name_liabilities"],
            options["name_equity"],
            options["name_income"],
            options["name_expenses"],
        )

    def join_path(self, *args: str) -> Path:
        """Path relative to the directory of the ledger."""
        return Path(self.beancount_file_path).parent.joinpath(*args).resolve()

    def paths_to_watch(self) -> tuple[Sequence[Path], Sequence[Path]]:
        """Get paths to included files and document directories.

        Returns:
            A tuple (files, directories).
        """
        files = [Path(i) for i in self.options["include"]]
        if self.ingest.module_path:
            files.append(self.ingest.module_path)
        return (
            files,
            [
                self.join_path(path, account)
                for account in self.root_accounts
                for path in self.options["documents"]
            ],
        )

    def changed(self) -> bool:
        """Check if the file needs to be reloaded.

        Returns:
            True if a change in one of the included files or a change in a
            document folder was detected and the file has been reloaded.
        """
        # We can't reload an encrypted file, so act like it never changes.
        if self._is_encrypted:  # pragma: no cover
            return False
        changed = self.watcher.check()
        if changed:
            self.load_file()
        return changed

    def interval_balances(
        self,
        filtered: FilteredLedger,
        interval: Interval,
        account_name: str,
        *,
        accumulate: bool = False,
    ) -> tuple[Sequence[Tree], Sequence[DateRange]]:
        """Balances by interval.

        Arguments:
            filtered: The currently filtered ledger.
            interval: An interval.
            account_name: An account name.
            accumulate: A boolean, ``True`` if the balances for an interval
                should include all entries up to the end of the interval.

        Returns:
            A pair of a list of Tree instances and the intervals.
        """
        min_accounts = [
            account
            for account in self.accounts
            if account.startswith(account_name)
        ]

        interval_ranges = list(reversed(filtered.interval_ranges(interval)))
        interval_balances = [
            Tree(
                slice_entry_dates(
                    filtered.entries,
                    date.min if accumulate else date_range.begin,
                    date_range.end,
                ),
                min_accounts,
            )
            for date_range in interval_ranges
        ]

        return interval_balances, interval_ranges

    @listify
    def account_journal(
        self,
        filtered: FilteredLedger,
        account_name: str,
        conversion: str | Conversion,
        *,
        with_children: bool,
    ) -> Iterable[
        tuple[int, Directive, SimpleCounterInventory, SimpleCounterInventory]
    ]:
        """Journal for an account.

        Args:
            filtered: The currently filtered ledger.
            account_name: An account name.
            conversion: The conversion to use.
            with_children: Whether to include postings of subaccounts of
                           the account.

        Yields:
            Tuples of ``(index, entry, change, balance)``.
        """
        conv = conversion_from_str(conversion)
        relevant_account = account_tester(
            account_name, with_children=with_children
        )

        prices = self.prices
        balance = CounterInventory()
        for index, entry in enumerate(filtered.entries_without_prices):
            change = CounterInventory()
            entry_is_relevant = False
            postings = getattr(entry, "postings", None)
            if postings is not None:
                for posting in postings:
                    if relevant_account(posting.account):
                        entry_is_relevant = True
                        balance.add_position(posting)
                        change.add_position(posting)
            elif any(relevant_account(a) for a in get_entry_accounts(entry)):
                entry_is_relevant = True

            if entry_is_relevant:
                yield (
                    index,
                    entry,
                    conv.apply(change, prices, entry.date),
                    conv.apply(balance, prices, entry.date),
                )

    def _get_entry(self, entry_hash: str) -> Directive:
        """Find an entry.

        Arguments:
            entry_hash: Hash of the entry.

        Returns:
            The entry with the given hash.

        Raises:
            EntryNotFoundForHashError: If there is no entry for the given hash.
        """
        try:
            return next(
                entry
                for entry in self.all_entries
                if entry_hash == hash_entry(entry)
            )
        except StopIteration as exc:
            raise EntryNotFoundForHashError(entry_hash) from exc

    def context(
        self,
        entry_hash: str,
    ) -> tuple[
        Directive,
        Mapping[str, Sequence[str]] | None,
        Mapping[str, Sequence[str]] | None,
    ]:
        """Context for an entry.

        Arguments:
            entry_hash: Hash of entry.

        Returns:
            A tuple ``(entry, before, after, source_slice, sha256sum)`` of the
            (unique) entry with the given ``entry_hash``. If the entry is a
            Balance or Transaction then ``before`` and ``after`` contain
            the balances before and after the entry of the affected accounts.
        """
        entry = self.get_entry(entry_hash)

        if not isinstance(entry, (Balance, Transaction)):
            return entry, None, None

        entry_accounts = get_entry_accounts(entry)
        balances = {account: CounterInventory() for account in entry_accounts}
        for entry_ in takewhile(lambda e: e is not entry, self.all_entries):
            if isinstance(entry_, Transaction):
                for posting in entry_.postings:
                    balance = balances.get(posting.account, None)
                    if balance is not None:
                        balance.add_position(posting)

        def visualise(inv: CounterInventory) -> Sequence[str]:
            return inv.to_strings()

        before = {acc: visualise(inv) for acc, inv in balances.items()}

        if isinstance(entry, Balance):
            return entry, before, None

        for posting in entry.postings:
            balances[posting.account].add_position(posting)
        after = {acc: visualise(inv) for acc, inv in balances.items()}
        return entry, before, after

    def commodity_pairs(self) -> Sequence[tuple[str, str]]:
        """List pairs of commodities.

        Returns:
            A list of pairs of commodities. Pairs of operating currencies will
            be given in both directions not just in the one found in file.
        """
        return self.prices.commodity_pairs(self.options["operating_currency"])

    def statement_path(self, entry_hash: str, metadata_key: str) -> str:
        """Get the path for a statement found in the specified entry.

        The entry that we look up should contain a path to a document (absolute
        or relative to the filename of the entry) or just its basename. We go
        through all documents and match on the full path or if one of the
        documents with a matching account has a matching file basename.

        Arguments:
            entry_hash: Hash of the entry containing the path in its metadata.
            metadata_key: The key that the path should be in.

        Returns:
            The filename of the matching document entry.

        Raises:
            StatementMetadataInvalidError: If the metadata at the given key is
                                           invalid.
            StatementNotFoundError: If no matching document is found.
        """
        entry = self.get_entry(entry_hash)
        value = entry.meta.get(metadata_key, None)
        if not isinstance(value, str):
            raise StatementMetadataInvalidError(metadata_key)

        accounts = set(get_entry_accounts(entry))
        filename, _ = get_position(entry)
        full_path = (Path(filename).parent / value).resolve()
        for document in self.all_entries_by_type.Document:
            document_path = Path(document.filename)
            if document_path == full_path:
                return document.filename
            if document.account in accounts and document_path.name == value:
                return document.filename

        raise StatementNotFoundError

    group_entries_by_type = staticmethod(group_entries_by_type)

mtime property

The timestamp to the latest change of the underlying files.

errors property

The errors that the Beancount loading plus Fava module errors.

root_accounts property

The five root accounts.

__init__(path, *, poll_watcher=False)

Create an interface for a Beancount ledger.

Parameters:

Name	Type	Description	Default
path	str	Path to the main Beancount file.	required
poll_watcher	bool	Whether to use the polling file watcher.	False

Source code in src/rustfava/core/__init__.py

def __init__(self, path: str, *, poll_watcher: bool = False) -> None:
    """Create an interface for a Beancount ledger.

    Arguments:
        path: Path to the main Beancount file.
        poll_watcher: Whether to use the polling file watcher.
    """
    #: The path to the main Beancount file.
    self.beancount_file_path = path
    self._is_encrypted = is_encrypted_file(path)
    self.get_filtered = lru_cache(maxsize=16)(self._get_filtered)
    self.get_entry = lru_cache(maxsize=16)(self._get_entry)

    self.accounts = AccountDict(self)
    self.attributes = AttributesModule(self)
    self.budgets = BudgetModule(self)
    self.charts = ChartModule(self)
    self.commodities = CommoditiesModule(self)
    self.extensions = ExtensionModule(self)
    self.file = FileModule(self)
    self.format_decimal = DecimalFormatModule(self)
    self.ingest = IngestModule(self)
    self.misc = FavaMisc(self)
    self.query_shell = QueryShell(self)

    self.watcher = WatchfilesWatcher() if not poll_watcher else Watcher()

    self.load_file()

load_file()

Load the main file and all included files and set attributes.

Source code in src/rustfava/core/__init__.py

def load_file(self) -> None:
    """Load the main file and all included files and set attributes."""
    self.all_entries, self.load_errors, self.options = load_uncached(
        self.beancount_file_path,
        is_encrypted=self._is_encrypted,
    )
    self.get_filtered.cache_clear()
    self.get_entry.cache_clear()

    self.all_entries_by_type = group_entries_by_type(self.all_entries)
    self.prices = RustfavaPriceMap(self.all_entries_by_type.Price)

    self.fava_options, self.fava_options_errors = parse_options(
        self.all_entries_by_type.Custom,
    )

    if self._is_encrypted:  # pragma: no cover
        pass
    else:
        self.watcher.update(*self.paths_to_watch())

    # Call load_file of all modules.
    self.accounts.load_file()
    self.attributes.load_file()
    self.budgets.load_file()
    self.charts.load_file()
    self.commodities.load_file()
    self.extensions.load_file()
    self.file.load_file()
    self.format_decimal.load_file()
    self.misc.load_file()
    self.query_shell.load_file()
    self.ingest.load_file()

    self.extensions.after_load_file()

join_path(*args)

Path relative to the directory of the ledger.

Source code in src/rustfava/core/__init__.py

def join_path(self, *args: str) -> Path:
    """Path relative to the directory of the ledger."""
    return Path(self.beancount_file_path).parent.joinpath(*args).resolve()

paths_to_watch()

Get paths to included files and document directories.

Returns:

Type	Description
tuple[Sequence[Path], Sequence[Path]]	A tuple (files, directories).

Source code in src/rustfava/core/__init__.py

def paths_to_watch(self) -> tuple[Sequence[Path], Sequence[Path]]:
    """Get paths to included files and document directories.

    Returns:
        A tuple (files, directories).
    """
    files = [Path(i) for i in self.options["include"]]
    if self.ingest.module_path:
        files.append(self.ingest.module_path)
    return (
        files,
        [
            self.join_path(path, account)
            for account in self.root_accounts
            for path in self.options["documents"]
        ],
    )

changed()

Check if the file needs to be reloaded.

Returns:

Type	Description
bool	True if a change in one of the included files or a change in a
bool	document folder was detected and the file has been reloaded.

Source code in src/rustfava/core/__init__.py

def changed(self) -> bool:
    """Check if the file needs to be reloaded.

    Returns:
        True if a change in one of the included files or a change in a
        document folder was detected and the file has been reloaded.
    """
    # We can't reload an encrypted file, so act like it never changes.
    if self._is_encrypted:  # pragma: no cover
        return False
    changed = self.watcher.check()
    if changed:
        self.load_file()
    return changed

interval_balances(filtered, interval, account_name, *, accumulate=False)

Balances by interval.

Parameters:

Name	Type	Description	Default
filtered	FilteredLedger	The currently filtered ledger.	required
interval	Interval	An interval.	required
account_name	str	An account name.	required
accumulate	bool	A boolean, True if the balances for an interval should include all entries up to the end of the interval.	False

Returns:

Type	Description
tuple[Sequence[Tree], Sequence[DateRange]]	A pair of a list of Tree instances and the intervals.

Source code in src/rustfava/core/__init__.py

def interval_balances(
    self,
    filtered: FilteredLedger,
    interval: Interval,
    account_name: str,
    *,
    accumulate: bool = False,
) -> tuple[Sequence[Tree], Sequence[DateRange]]:
    """Balances by interval.

    Arguments:
        filtered: The currently filtered ledger.
        interval: An interval.
        account_name: An account name.
        accumulate: A boolean, ``True`` if the balances for an interval
            should include all entries up to the end of the interval.

    Returns:
        A pair of a list of Tree instances and the intervals.
    """
    min_accounts = [
        account
        for account in self.accounts
        if account.startswith(account_name)
    ]

    interval_ranges = list(reversed(filtered.interval_ranges(interval)))
    interval_balances = [
        Tree(
            slice_entry_dates(
                filtered.entries,
                date.min if accumulate else date_range.begin,
                date_range.end,
            ),
            min_accounts,
        )
        for date_range in interval_ranges
    ]

    return interval_balances, interval_ranges

account_journal(filtered, account_name, conversion, *, with_children)

Journal for an account.

Parameters:

Name	Type	Description	Default
filtered	FilteredLedger	The currently filtered ledger.	required
account_name	str	An account name.	required
conversion	str | Conversion	The conversion to use.	required
with_children	bool	Whether to include postings of subaccounts of the account.	required

Yields:

Type	Description
Iterable[tuple[int, Directive, SimpleCounterInventory, SimpleCounterInventory]]	Tuples of (index, entry, change, balance).

Source code in src/rustfava/core/__init__.py

@listify
def account_journal(
    self,
    filtered: FilteredLedger,
    account_name: str,
    conversion: str | Conversion,
    *,
    with_children: bool,
) -> Iterable[
    tuple[int, Directive, SimpleCounterInventory, SimpleCounterInventory]
]:
    """Journal for an account.

    Args:
        filtered: The currently filtered ledger.
        account_name: An account name.
        conversion: The conversion to use.
        with_children: Whether to include postings of subaccounts of
                       the account.

    Yields:
        Tuples of ``(index, entry, change, balance)``.
    """
    conv = conversion_from_str(conversion)
    relevant_account = account_tester(
        account_name, with_children=with_children
    )

    prices = self.prices
    balance = CounterInventory()
    for index, entry in enumerate(filtered.entries_without_prices):
        change = CounterInventory()
        entry_is_relevant = False
        postings = getattr(entry, "postings", None)
        if postings is not None:
            for posting in postings:
                if relevant_account(posting.account):
                    entry_is_relevant = True
                    balance.add_position(posting)
                    change.add_position(posting)
        elif any(relevant_account(a) for a in get_entry_accounts(entry)):
            entry_is_relevant = True

        if entry_is_relevant:
            yield (
                index,
                entry,
                conv.apply(change, prices, entry.date),
                conv.apply(balance, prices, entry.date),
            )

context(entry_hash)

Context for an entry.

Parameters:

Name	Type	Description	Default
entry_hash	str	Hash of entry.	required

Returns:

Type	Description
Directive	A tuple (entry, before, after, source_slice, sha256sum) of the
Mapping[str, Sequence[str]] | None	(unique) entry with the given entry_hash. If the entry is a
Mapping[str, Sequence[str]] | None	Balance or Transaction then before and after contain
tuple[Directive, Mapping[str, Sequence[str]] | None, Mapping[str, Sequence[str]] | None]	the balances before and after the entry of the affected accounts.

Source code in src/rustfava/core/__init__.py

def context(
    self,
    entry_hash: str,
) -> tuple[
    Directive,
    Mapping[str, Sequence[str]] | None,
    Mapping[str, Sequence[str]] | None,
]:
    """Context for an entry.

    Arguments:
        entry_hash: Hash of entry.

    Returns:
        A tuple ``(entry, before, after, source_slice, sha256sum)`` of the
        (unique) entry with the given ``entry_hash``. If the entry is a
        Balance or Transaction then ``before`` and ``after`` contain
        the balances before and after the entry of the affected accounts.
    """
    entry = self.get_entry(entry_hash)

    if not isinstance(entry, (Balance, Transaction)):
        return entry, None, None

    entry_accounts = get_entry_accounts(entry)
    balances = {account: CounterInventory() for account in entry_accounts}
    for entry_ in takewhile(lambda e: e is not entry, self.all_entries):
        if isinstance(entry_, Transaction):
            for posting in entry_.postings:
                balance = balances.get(posting.account, None)
                if balance is not None:
                    balance.add_position(posting)

    def visualise(inv: CounterInventory) -> Sequence[str]:
        return inv.to_strings()

    before = {acc: visualise(inv) for acc, inv in balances.items()}

    if isinstance(entry, Balance):
        return entry, before, None

    for posting in entry.postings:
        balances[posting.account].add_position(posting)
    after = {acc: visualise(inv) for acc, inv in balances.items()}
    return entry, before, after

commodity_pairs()

List pairs of commodities.

Returns:

Type	Description
Sequence[tuple[str, str]]	A list of pairs of commodities. Pairs of operating currencies will
Sequence[tuple[str, str]]	be given in both directions not just in the one found in file.

Source code in src/rustfava/core/__init__.py

def commodity_pairs(self) -> Sequence[tuple[str, str]]:
    """List pairs of commodities.

    Returns:
        A list of pairs of commodities. Pairs of operating currencies will
        be given in both directions not just in the one found in file.
    """
    return self.prices.commodity_pairs(self.options["operating_currency"])

statement_path(entry_hash, metadata_key)

Get the path for a statement found in the specified entry.

The entry that we look up should contain a path to a document (absolute or relative to the filename of the entry) or just its basename. We go through all documents and match on the full path or if one of the documents with a matching account has a matching file basename.

Parameters:

Name	Type	Description	Default
entry_hash	str	Hash of the entry containing the path in its metadata.	required
metadata_key	str	The key that the path should be in.	required

Returns:

Type	Description
str	The filename of the matching document entry.

Raises:

Type	Description
StatementMetadataInvalidError	If the metadata at the given key is invalid.
StatementNotFoundError	If no matching document is found.

Source code in src/rustfava/core/__init__.py

def statement_path(self, entry_hash: str, metadata_key: str) -> str:
    """Get the path for a statement found in the specified entry.

    The entry that we look up should contain a path to a document (absolute
    or relative to the filename of the entry) or just its basename. We go
    through all documents and match on the full path or if one of the
    documents with a matching account has a matching file basename.

    Arguments:
        entry_hash: Hash of the entry containing the path in its metadata.
        metadata_key: The key that the path should be in.

    Returns:
        The filename of the matching document entry.

    Raises:
        StatementMetadataInvalidError: If the metadata at the given key is
                                       invalid.
        StatementNotFoundError: If no matching document is found.
    """
    entry = self.get_entry(entry_hash)
    value = entry.meta.get(metadata_key, None)
    if not isinstance(value, str):
        raise StatementMetadataInvalidError(metadata_key)

    accounts = set(get_entry_accounts(entry))
    filename, _ = get_position(entry)
    full_path = (Path(filename).parent / value).resolve()
    for document in self.all_entries_by_type.Document:
        document_path = Path(document.filename)
        if document_path == full_path:
            return document.filename
        if document.account in accounts and document_path.name == value:
            return document.filename

    raise StatementNotFoundError

accounts

Account close date and metadata.

LastEntry dataclass

Date and hash of the last entry for an account.

Source code in src/rustfava/core/accounts.py

@dataclass(frozen=True)
class LastEntry:
    """Date and hash of the last entry for an account."""

    #: The entry date.
    date: datetime.date

    #: The entry hash.
    entry_hash: str

AccountData dataclass

Holds information about an account.

Source code in src/rustfava/core/accounts.py

@dataclass
class AccountData:
    """Holds information about an account."""

    #: The date on which this account is closed (or datetime.date.max).
    close_date: datetime.date | None = None

    #: The metadata of the Open entry of this account.
    meta: Meta = field(default_factory=dict)

    #: Uptodate status. Is only computed if the account has a
    #: "fava-uptodate-indication" meta attribute.
    uptodate_status: Literal["green", "yellow", "red"] | None = None

    #: Balance directive if this account has an uptodate status.
    balance_string: str | None = None

    #: The last entry of the account (unless it is a close Entry)
    last_entry: LastEntry | None = None

AccountDict

Bases: FavaModule, dict[str, AccountData]

Account info dictionary.

Source code in src/rustfava/core/accounts.py

class AccountDict(FavaModule, dict[str, AccountData]):
    """Account info dictionary."""

    EMPTY = AccountData()

    def __missing__(self, key: str) -> AccountData:
        return self.EMPTY

    def setdefault(
        self,
        key: str,
        _: AccountData | None = None,
    ) -> AccountData:
        """Get the account of the given name, insert one if it is missing."""
        if key not in self:
            self[key] = AccountData()
        return self[key]

    def load_file(self) -> None:  # noqa: D102
        self.clear()
        entries_by_account = group_entries_by_account(self.ledger.all_entries)
        tree = Tree(self.ledger.all_entries)
        for open_entry in self.ledger.all_entries_by_type.Open:
            meta = open_entry.meta
            account_data = self.setdefault(open_entry.account)
            account_data.meta = meta

            txn_postings = entries_by_account[open_entry.account]
            last = get_last_entry(txn_postings)
            if last is not None and not isinstance(last, Close):
                account_data.last_entry = LastEntry(
                    date=last.date,
                    entry_hash=hash_entry(last),
                )
            if meta.get("fava-uptodate-indication"):
                account_data.uptodate_status = uptodate_status(txn_postings)
                if account_data.uptodate_status != "green":
                    account_data.balance_string = balance_string(
                        tree.get(open_entry.account),
                    )
        for close in self.ledger.all_entries_by_type.Close:
            self.setdefault(close.account).close_date = close.date

    def all_balance_directives(self) -> str:
        """Balance directives for all accounts."""
        return "".join(
            account_details.balance_string
            for account_details in self.values()
            if account_details.balance_string
        )

setdefault(key, _=None)

Get the account of the given name, insert one if it is missing.

Source code in src/rustfava/core/accounts.py

def setdefault(
    self,
    key: str,
    _: AccountData | None = None,
) -> AccountData:
    """Get the account of the given name, insert one if it is missing."""
    if key not in self:
        self[key] = AccountData()
    return self[key]

all_balance_directives()

Balance directives for all accounts.

Source code in src/rustfava/core/accounts.py

def all_balance_directives(self) -> str:
    """Balance directives for all accounts."""
    return "".join(
        account_details.balance_string
        for account_details in self.values()
        if account_details.balance_string
    )

get_last_entry(txn_postings)

Last entry.

Source code in src/rustfava/core/accounts.py

def get_last_entry(
    txn_postings: Sequence[Directive | TransactionPosting],
) -> Directive | None:
    """Last entry."""
    for txn_posting in reversed(txn_postings):
        if isinstance(txn_posting, TransactionPosting):
            transaction = txn_posting.transaction
            if transaction.flag != FLAG_UNREALIZED:
                return transaction
        else:
            return txn_posting
    return None

uptodate_status(txn_postings)

Status of the last balance or transaction.

Parameters:

Name	Type	Description	Default
txn_postings	Sequence[Directive | TransactionPosting]	The TransactionPosting for the account.	required

Returns:

Type	Description
Literal['green', 'yellow', 'red'] | None	A status string for the last balance or transaction of the account.
Literal['green', 'yellow', 'red'] | None	'green': A balance check that passed.
Literal['green', 'yellow', 'red'] | None	'red': A balance check that failed.
Literal['green', 'yellow', 'red'] | None	'yellow': Not a balance check.

Source code in src/rustfava/core/accounts.py

def uptodate_status(
    txn_postings: Sequence[Directive | TransactionPosting],
) -> Literal["green", "yellow", "red"] | None:
    """Status of the last balance or transaction.

    Args:
        txn_postings: The TransactionPosting for the account.

    Returns:
        A status string for the last balance or transaction of the account.

        - 'green':  A balance check that passed.
        - 'red':    A balance check that failed.
        - 'yellow': Not a balance check.
    """
    for txn_posting in reversed(txn_postings):
        if isinstance(txn_posting, Balance):
            return "red" if txn_posting.diff_amount else "green"
        if (
            isinstance(txn_posting, TransactionPosting)
            and txn_posting.transaction.flag != FLAG_UNREALIZED
        ):
            return "yellow"
    return None

balance_string(tree_node)

Balance directive for the given account for today.

Source code in src/rustfava/core/accounts.py

def balance_string(tree_node: TreeNode) -> str:
    """Balance directive for the given account for today."""
    account = tree_node.name
    today = str(local_today())
    res = ""
    for currency, number in UNITS.apply(tree_node.balance).items():
        res += f"{today} balance {account:<28} {number:>15} {currency}\n"
    return res

attributes

Attributes for auto-completion.

AttributesModule

Bases: FavaModule

Some attributes of the ledger (mostly for auto-completion).

Source code in src/rustfava/core/attributes.py

class AttributesModule(FavaModule):
    """Some attributes of the ledger (mostly for auto-completion)."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        self.accounts: Sequence[str] = []
        self.currencies: Sequence[str] = []
        self.payees: Sequence[str] = []
        self.links: Sequence[str] = []
        self.tags: Sequence[str] = []
        self.years: Sequence[str] = []

    def load_file(self) -> None:  # noqa: D102
        all_entries = self.ledger.all_entries

        all_links = set()
        all_tags = set()
        for entry in all_entries:
            links = getattr(entry, "links", None)
            if links is not None:
                all_links.update(links)
            tags = getattr(entry, "tags", None)
            if tags is not None:
                all_tags.update(tags)
        self.links = sorted(all_links)
        self.tags = sorted(all_tags)

        self.years = get_active_years(
            all_entries,
            self.ledger.fava_options.fiscal_year_end,
        )

        account_ranker = ExponentialDecayRanker(
            sorted(self.ledger.accounts.keys()),
        )
        currency_ranker = ExponentialDecayRanker()
        payee_ranker = ExponentialDecayRanker()

        for txn in self.ledger.all_entries_by_type.Transaction:
            if txn.payee:
                payee_ranker.update(txn.payee, txn.date)
            for posting in txn.postings:
                account_ranker.update(posting.account, txn.date)
                # Skip postings with missing units (can happen with parse errors)
                if posting.units is not None:
                    currency_ranker.update(posting.units.currency, txn.date)
                if posting.cost and posting.cost.currency is not None:
                    currency_ranker.update(posting.cost.currency, txn.date)

        self.accounts = account_ranker.sort()
        self.currencies = currency_ranker.sort()
        self.payees = payee_ranker.sort()

    def payee_accounts(self, payee: str) -> Sequence[str]:
        """Rank accounts for the given payee."""
        account_ranker = ExponentialDecayRanker(self.accounts)
        transactions = self.ledger.all_entries_by_type.Transaction
        for txn in transactions:
            if txn.payee == payee:
                for posting in txn.postings:
                    account_ranker.update(posting.account, txn.date)
        return account_ranker.sort()

    def payee_transaction(self, payee: str) -> Transaction | None:
        """Get the last transaction for a payee."""
        transactions = self.ledger.all_entries_by_type.Transaction
        for txn in reversed(transactions):
            if txn.payee == payee:
                return txn
        return None

    def narration_transaction(self, narration: str) -> Transaction | None:
        """Get the last transaction for a narration."""
        transactions = self.ledger.all_entries_by_type.Transaction
        for txn in reversed(transactions):
            if txn.narration == narration:
                return txn
        return None

    @property
    def narrations(self) -> Sequence[str]:
        """Get the narrations of all transactions."""
        narration_ranker = ExponentialDecayRanker()
        for txn in self.ledger.all_entries_by_type.Transaction:
            if txn.narration:
                narration_ranker.update(txn.narration, txn.date)
        return narration_ranker.sort()

narrations property

Get the narrations of all transactions.

payee_accounts(payee)

Rank accounts for the given payee.

Source code in src/rustfava/core/attributes.py

def payee_accounts(self, payee: str) -> Sequence[str]:
    """Rank accounts for the given payee."""
    account_ranker = ExponentialDecayRanker(self.accounts)
    transactions = self.ledger.all_entries_by_type.Transaction
    for txn in transactions:
        if txn.payee == payee:
            for posting in txn.postings:
                account_ranker.update(posting.account, txn.date)
    return account_ranker.sort()

payee_transaction(payee)

Get the last transaction for a payee.

Source code in src/rustfava/core/attributes.py

def payee_transaction(self, payee: str) -> Transaction | None:
    """Get the last transaction for a payee."""
    transactions = self.ledger.all_entries_by_type.Transaction
    for txn in reversed(transactions):
        if txn.payee == payee:
            return txn
    return None

narration_transaction(narration)

Get the last transaction for a narration.

Source code in src/rustfava/core/attributes.py

def narration_transaction(self, narration: str) -> Transaction | None:
    """Get the last transaction for a narration."""
    transactions = self.ledger.all_entries_by_type.Transaction
    for txn in reversed(transactions):
        if txn.narration == narration:
            return txn
    return None

get_active_years(entries, fye)

Return active years, with support for fiscal years.

Parameters:

Name	Type	Description	Default
entries	Sequence[Directive]	Beancount entries	required
fye	FiscalYearEnd	fiscal year end	required

Returns:

Type	Description
list[str]	A reverse sorted list of years or fiscal years that occur in the
list[str]	entries.

Source code in src/rustfava/core/attributes.py

def get_active_years(
    entries: Sequence[Directive],
    fye: FiscalYearEnd,
) -> list[str]:
    """Return active years, with support for fiscal years.

    Args:
        entries: Beancount entries
        fye: fiscal year end

    Returns:
        A reverse sorted list of years or fiscal years that occur in the
        entries.
    """
    years = []
    if fye == END_OF_YEAR:
        prev_year = None
        for entry in entries:
            year = entry.date.year
            if year != prev_year:
                prev_year = year
                years.append(year)
        return [f"{year}" for year in reversed(years)]
    month = fye.month
    day = fye.day
    prev_year = None
    for entry in entries:
        date = entry.date
        year = (
            entry.date.year + 1
            if date.month > month or (date.month == month and date.day > day)
            else entry.date.year
        )
        if year != prev_year:
            prev_year = year
            years.append(year)
    return [f"FY{year}" for year in reversed(years)]

budgets

Parsing and computing budgets.

BudgetDict = dict[str, list[Budget]] module-attribute

A map of account names to lists of budget entries.

Budget

Bases: NamedTuple

A budget entry.

Source code in src/rustfava/core/budgets.py

class Budget(NamedTuple):
    """A budget entry."""

    account: str
    date_start: datetime.date
    period: Interval
    number: Decimal
    currency: str

BudgetError dataclass

Bases: BeancountError

Error with a budget.

Source code in src/rustfava/core/budgets.py

class BudgetError(BeancountError):
    """Error with a budget."""

BudgetModule

Bases: FavaModule

Parses budget entries.

Source code in src/rustfava/core/budgets.py

class BudgetModule(FavaModule):
    """Parses budget entries."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        self._budget_entries: BudgetDict = {}
        self.errors: Sequence[BudgetError] = []

    def load_file(self) -> None:  # noqa: D102
        self._budget_entries, self.errors = parse_budgets(
            self.ledger.all_entries_by_type.Custom,
        )

    def calculate(
        self,
        account: str,
        begin_date: datetime.date,
        end_date: datetime.date,
    ) -> Mapping[str, Decimal]:
        """Calculate the budget for an account in an interval."""
        return calculate_budget(
            self._budget_entries,
            account,
            begin_date,
            end_date,
        )

    def calculate_children(
        self,
        account: str,
        begin_date: datetime.date,
        end_date: datetime.date,
    ) -> Mapping[str, Decimal]:
        """Calculate the budget for an account including its children."""
        return calculate_budget_children(
            self._budget_entries,
            account,
            begin_date,
            end_date,
        )

calculate(account, begin_date, end_date)

Calculate the budget for an account in an interval.

Source code in src/rustfava/core/budgets.py

def calculate(
    self,
    account: str,
    begin_date: datetime.date,
    end_date: datetime.date,
) -> Mapping[str, Decimal]:
    """Calculate the budget for an account in an interval."""
    return calculate_budget(
        self._budget_entries,
        account,
        begin_date,
        end_date,
    )

calculate_children(account, begin_date, end_date)

Calculate the budget for an account including its children.

Source code in src/rustfava/core/budgets.py

def calculate_children(
    self,
    account: str,
    begin_date: datetime.date,
    end_date: datetime.date,
) -> Mapping[str, Decimal]:
    """Calculate the budget for an account including its children."""
    return calculate_budget_children(
        self._budget_entries,
        account,
        begin_date,
        end_date,
    )

parse_budgets(custom_entries)

Parse budget directives from custom entries.

Parameters:

Name	Type	Description	Default
custom_entries	Sequence[Custom]	the Custom entries to parse budgets from.	required

Returns:

Type	Description
tuple[BudgetDict, Sequence[BudgetError]]	A dict of accounts to lists of budgets.

Example

2015-04-09 custom "budget" Expenses:Books "monthly" 20.00 EUR

Source code in src/rustfava/core/budgets.py

def parse_budgets(
    custom_entries: Sequence[Custom],
) -> tuple[BudgetDict, Sequence[BudgetError]]:
    """Parse budget directives from custom entries.

    Args:
        custom_entries: the Custom entries to parse budgets from.

    Returns:
        A dict of accounts to lists of budgets.

    Example:
        2015-04-09 custom "budget" Expenses:Books "monthly" 20.00 EUR
    """
    budgets: BudgetDict = defaultdict(list)
    errors = []

    for entry in (entry for entry in custom_entries if entry.type == "budget"):
        try:
            interval = INTERVALS.get(str(entry.values[1].value).lower())
            if not interval:
                errors.append(
                    BudgetError(
                        entry.meta,
                        "Invalid interval for budget entry",
                        entry,
                    ),
                )
                continue
            budget = Budget(
                entry.values[0].value,
                entry.date,
                interval,
                entry.values[2].value.number,
                entry.values[2].value.currency,
            )
            budgets[budget.account].append(budget)
        except (IndexError, TypeError):
            errors.append(
                BudgetError(entry.meta, "Failed to parse budget entry", entry),
            )

    return budgets, errors

calculate_budget(budgets, account, date_from, date_to)

Calculate budget for an account.

Parameters:

Name	Type	Description	Default
budgets	BudgetDict	A list of :class:Budget entries.	required
account	str	An account name.	required
date_from	date	Starting date.	required
date_to	date	End date (exclusive).	required

Returns:

Type	Description
Mapping[str, Decimal]	A dictionary of currency to Decimal with the budget for the
Mapping[str, Decimal]	specified account and period.

Source code in src/rustfava/core/budgets.py

def calculate_budget(
    budgets: BudgetDict,
    account: str,
    date_from: datetime.date,
    date_to: datetime.date,
) -> Mapping[str, Decimal]:
    """Calculate budget for an account.

    Args:
        budgets: A list of :class:`Budget` entries.
        account: An account name.
        date_from: Starting date.
        date_to: End date (exclusive).

    Returns:
        A dictionary of currency to Decimal with the budget for the
        specified account and period.
    """
    budget_list = budgets.get(account, None)
    if budget_list is None:
        return {}

    currency_dict: dict[str, Decimal] = defaultdict(Decimal)

    for day in days_in_daterange(date_from, date_to):
        matches = _matching_budgets(budget_list, day)
        for budget in matches.values():
            days_in_period = budget.period.number_of_days(day)
            currency_dict[budget.currency] += budget.number / days_in_period
    return dict(currency_dict)

calculate_budget_children(budgets, account, date_from, date_to)

Calculate budget for an account including budgets of its children.

Parameters:

Name	Type	Description	Default
budgets	BudgetDict	A list of :class:Budget entries.	required
account	str	An account name.	required
date_from	date	Starting date.	required
date_to	date	End date (exclusive).	required

Returns:

Type	Description
Mapping[str, Decimal]	A dictionary of currency to Decimal with the budget for the
Mapping[str, Decimal]	specified account and period.

Source code in src/rustfava/core/budgets.py

def calculate_budget_children(
    budgets: BudgetDict,
    account: str,
    date_from: datetime.date,
    date_to: datetime.date,
) -> Mapping[str, Decimal]:
    """Calculate budget for an account including budgets of its children.

    Args:
        budgets: A list of :class:`Budget` entries.
        account: An account name.
        date_from: Starting date.
        date_to: End date (exclusive).

    Returns:
        A dictionary of currency to Decimal with the budget for the
        specified account and period.
    """
    currency_dict: dict[str, Decimal] = Counter()  # type: ignore[assignment]

    for child in budgets:
        if child.startswith(account):
            currency_dict.update(
                calculate_budget(budgets, child, date_from, date_to),
            )
    return dict(currency_dict)

charts

Provide data suitable for rustfava's charts.

RustfavaJSONProvider

Bases: JSONProvider

Use custom JSON encoder and decoder.

Source code in src/rustfava/core/charts.py

class RustfavaJSONProvider(JSONProvider):
    """Use custom JSON encoder and decoder."""

    def dumps(self, obj: Any, **_kwargs: Any) -> str:  # noqa: D102
        return json.dumps(
            obj, sort_keys=True, separators=(",", ":"), default=_json_default
        )

    def loads(self, s: str | bytes, **_kwargs: Any) -> Any:  # noqa: D102
        return json.loads(s)

DateAndBalance dataclass

Balance at a date.

Source code in src/rustfava/core/charts.py

@dataclass(frozen=True)
class DateAndBalance:
    """Balance at a date."""

    date: date
    balance: SimpleCounterInventory

DateAndBalanceWithBudget dataclass

Balance at a date with a budget.

Source code in src/rustfava/core/charts.py

@dataclass(frozen=True)
class DateAndBalanceWithBudget:
    """Balance at a date with a budget."""

    date: date
    balance: SimpleCounterInventory
    account_balances: Mapping[str, SimpleCounterInventory]
    budgets: Mapping[str, Decimal]

ChartModule

Bases: FavaModule

Return data for the various charts in rustfava.

Source code in src/rustfava/core/charts.py

class ChartModule(FavaModule):
    """Return data for the various charts in rustfava."""

    def hierarchy(
        self,
        filtered: FilteredLedger,
        account_name: str,
        conversion: Conversion,
    ) -> SerialisedTreeNode:
        """Render an account tree."""
        tree = filtered.root_tree
        return tree.get(account_name).serialise(
            conversion, self.ledger.prices, filtered.end_date
        )

    @listify
    def interval_totals(
        self,
        filtered: FilteredLedger,
        interval: Interval,
        accounts: str | tuple[str, ...],
        conversion: str | Conversion,
        *,
        invert: bool = False,
    ) -> Iterable[DateAndBalanceWithBudget]:
        """Render totals for account (or accounts) in the intervals.

        Args:
            filtered: The filtered ledger.
            interval: An interval.
            accounts: A single account (str) or a tuple of accounts.
            conversion: The conversion to use.
            invert: invert all numbers.

        Yields:
            The balances and budgets for the intervals.
        """
        conv = conversion_from_str(conversion)
        prices = self.ledger.prices

        # limit the bar charts to 100 intervals
        intervals = filtered.interval_ranges(interval)[-100:]

        for date_range in intervals:
            inventory = CounterInventory()
            entries = slice_entry_dates(
                filtered.entries, date_range.begin, date_range.end
            )
            account_inventories: dict[str, CounterInventory] = defaultdict(
                CounterInventory,
            )
            for entry in entries:
                for posting in getattr(entry, "postings", []):
                    if posting.account.startswith(accounts):
                        account_inventories[posting.account].add_position(
                            posting,
                        )
                        inventory.add_position(posting)
            balance = conv.apply(
                inventory,
                prices,
                date_range.end_inclusive,
            )
            account_balances = {
                account: conv.apply(
                    acct_value,
                    prices,
                    date_range.end_inclusive,
                )
                for account, acct_value in account_inventories.items()
            }
            budgets = (
                self.ledger.budgets.calculate_children(
                    accounts,
                    date_range.begin,
                    date_range.end,
                )
                if isinstance(accounts, str)
                else {}
            )

            if invert:
                balance = -balance
                budgets = {k: -v for k, v in budgets.items()}
                account_balances = {k: -v for k, v in account_balances.items()}

            yield DateAndBalanceWithBudget(
                date_range.end_inclusive,
                balance,
                account_balances,
                budgets,
            )

    @listify
    def linechart(
        self,
        filtered: FilteredLedger,
        account_name: str,
        conversion: str | Conversion,
    ) -> Iterable[DateAndBalance]:
        """Get the balance of an account as a line chart.

        Args:
            filtered: The filtered ledger.
            account_name: A string.
            conversion: The conversion to use.

        Yields:
            Dicts for all dates on which the balance of the given
            account has changed containing the balance (in units) of the
            account at that date.
        """
        conv = conversion_from_str(conversion)

        def _balances() -> Iterable[tuple[date, CounterInventory]]:
            last_date = None
            running_balance = CounterInventory()
            is_child_account = account_tester(account_name, with_children=True)

            for entry in filtered.entries:
                for posting in getattr(entry, "postings", []):
                    if is_child_account(posting.account):
                        new_date = entry.date
                        if last_date is not None and new_date > last_date:
                            yield (last_date, running_balance)
                        running_balance.add_position(posting)
                        last_date = new_date

            if last_date is not None:
                yield (last_date, running_balance)

        # When the balance for a commodity just went to zero, it will be
        # missing from the 'balance' so keep track of currencies that last had
        # a balance.
        last_currencies = None
        prices = self.ledger.prices

        for d, running_bal in _balances():
            balance = conv.apply(running_bal, prices, d)
            currencies = set(balance.keys())
            if last_currencies:
                for currency in last_currencies - currencies:
                    balance[currency] = ZERO
            last_currencies = currencies
            yield DateAndBalance(d, balance)

    @listify
    def net_worth(
        self,
        filtered: FilteredLedger,
        interval: Interval,
        conversion: str | Conversion,
    ) -> Iterable[DateAndBalance]:
        """Compute net worth.

        Args:
            filtered: The filtered ledger.
            interval: A string for the interval.
            conversion: The conversion to use.

        Yields:
            Dicts for all ends of the given interval containing the
            net worth (Assets + Liabilities) separately converted to all
            operating currencies.
        """
        conv = conversion_from_str(conversion)
        transactions = (
            entry
            for entry in filtered.entries
            if (
                isinstance(entry, Transaction)
                and entry.flag != FLAG_UNREALIZED
            )
        )

        types = (
            self.ledger.options["name_assets"],
            self.ledger.options["name_liabilities"],
        )

        txn = next(transactions, None)
        inventory = CounterInventory()

        prices = self.ledger.prices
        for date_range in filtered.interval_ranges(interval):
            while txn and txn.date < date_range.end:
                for posting in txn.postings:
                    if posting.account.startswith(types):
                        inventory.add_position(posting)
                txn = next(transactions, None)
            yield DateAndBalance(
                date_range.end_inclusive,
                conv.apply(
                    inventory,
                    prices,
                    date_range.end_inclusive,
                ),
            )

hierarchy(filtered, account_name, conversion)

Render an account tree.

Source code in src/rustfava/core/charts.py

def hierarchy(
    self,
    filtered: FilteredLedger,
    account_name: str,
    conversion: Conversion,
) -> SerialisedTreeNode:
    """Render an account tree."""
    tree = filtered.root_tree
    return tree.get(account_name).serialise(
        conversion, self.ledger.prices, filtered.end_date
    )

interval_totals(filtered, interval, accounts, conversion, *, invert=False)

Render totals for account (or accounts) in the intervals.

Parameters:

Name	Type	Description	Default
filtered	FilteredLedger	The filtered ledger.	required
interval	Interval	An interval.	required
accounts	str | tuple[str, ...]	A single account (str) or a tuple of accounts.	required
conversion	str | Conversion	The conversion to use.	required
invert	bool	invert all numbers.	False

Yields:

Type	Description
Iterable[DateAndBalanceWithBudget]	The balances and budgets for the intervals.

Source code in src/rustfava/core/charts.py

@listify
def interval_totals(
    self,
    filtered: FilteredLedger,
    interval: Interval,
    accounts: str | tuple[str, ...],
    conversion: str | Conversion,
    *,
    invert: bool = False,
) -> Iterable[DateAndBalanceWithBudget]:
    """Render totals for account (or accounts) in the intervals.

    Args:
        filtered: The filtered ledger.
        interval: An interval.
        accounts: A single account (str) or a tuple of accounts.
        conversion: The conversion to use.
        invert: invert all numbers.

    Yields:
        The balances and budgets for the intervals.
    """
    conv = conversion_from_str(conversion)
    prices = self.ledger.prices

    # limit the bar charts to 100 intervals
    intervals = filtered.interval_ranges(interval)[-100:]

    for date_range in intervals:
        inventory = CounterInventory()
        entries = slice_entry_dates(
            filtered.entries, date_range.begin, date_range.end
        )
        account_inventories: dict[str, CounterInventory] = defaultdict(
            CounterInventory,
        )
        for entry in entries:
            for posting in getattr(entry, "postings", []):
                if posting.account.startswith(accounts):
                    account_inventories[posting.account].add_position(
                        posting,
                    )
                    inventory.add_position(posting)
        balance = conv.apply(
            inventory,
            prices,
            date_range.end_inclusive,
        )
        account_balances = {
            account: conv.apply(
                acct_value,
                prices,
                date_range.end_inclusive,
            )
            for account, acct_value in account_inventories.items()
        }
        budgets = (
            self.ledger.budgets.calculate_children(
                accounts,
                date_range.begin,
                date_range.end,
            )
            if isinstance(accounts, str)
            else {}
        )

        if invert:
            balance = -balance
            budgets = {k: -v for k, v in budgets.items()}
            account_balances = {k: -v for k, v in account_balances.items()}

        yield DateAndBalanceWithBudget(
            date_range.end_inclusive,
            balance,
            account_balances,
            budgets,
        )

linechart(filtered, account_name, conversion)

Get the balance of an account as a line chart.

Parameters:

Name	Type	Description	Default
filtered	FilteredLedger	The filtered ledger.	required
account_name	str	A string.	required
conversion	str | Conversion	The conversion to use.	required

Yields:

Type	Description
Iterable[DateAndBalance]	Dicts for all dates on which the balance of the given
Iterable[DateAndBalance]	account has changed containing the balance (in units) of the
Iterable[DateAndBalance]	account at that date.

Source code in src/rustfava/core/charts.py

@listify
def linechart(
    self,
    filtered: FilteredLedger,
    account_name: str,
    conversion: str | Conversion,
) -> Iterable[DateAndBalance]:
    """Get the balance of an account as a line chart.

    Args:
        filtered: The filtered ledger.
        account_name: A string.
        conversion: The conversion to use.

    Yields:
        Dicts for all dates on which the balance of the given
        account has changed containing the balance (in units) of the
        account at that date.
    """
    conv = conversion_from_str(conversion)

    def _balances() -> Iterable[tuple[date, CounterInventory]]:
        last_date = None
        running_balance = CounterInventory()
        is_child_account = account_tester(account_name, with_children=True)

        for entry in filtered.entries:
            for posting in getattr(entry, "postings", []):
                if is_child_account(posting.account):
                    new_date = entry.date
                    if last_date is not None and new_date > last_date:
                        yield (last_date, running_balance)
                    running_balance.add_position(posting)
                    last_date = new_date

        if last_date is not None:
            yield (last_date, running_balance)

    # When the balance for a commodity just went to zero, it will be
    # missing from the 'balance' so keep track of currencies that last had
    # a balance.
    last_currencies = None
    prices = self.ledger.prices

    for d, running_bal in _balances():
        balance = conv.apply(running_bal, prices, d)
        currencies = set(balance.keys())
        if last_currencies:
            for currency in last_currencies - currencies:
                balance[currency] = ZERO
        last_currencies = currencies
        yield DateAndBalance(d, balance)

net_worth(filtered, interval, conversion)

Compute net worth.

Parameters:

Name	Type	Description	Default
filtered	FilteredLedger	The filtered ledger.	required
interval	Interval	A string for the interval.	required
conversion	str | Conversion	The conversion to use.	required

Yields:

Type	Description
Iterable[DateAndBalance]	Dicts for all ends of the given interval containing the
Iterable[DateAndBalance]	net worth (Assets + Liabilities) separately converted to all
Iterable[DateAndBalance]	operating currencies.

Source code in src/rustfava/core/charts.py

@listify
def net_worth(
    self,
    filtered: FilteredLedger,
    interval: Interval,
    conversion: str | Conversion,
) -> Iterable[DateAndBalance]:
    """Compute net worth.

    Args:
        filtered: The filtered ledger.
        interval: A string for the interval.
        conversion: The conversion to use.

    Yields:
        Dicts for all ends of the given interval containing the
        net worth (Assets + Liabilities) separately converted to all
        operating currencies.
    """
    conv = conversion_from_str(conversion)
    transactions = (
        entry
        for entry in filtered.entries
        if (
            isinstance(entry, Transaction)
            and entry.flag != FLAG_UNREALIZED
        )
    )

    types = (
        self.ledger.options["name_assets"],
        self.ledger.options["name_liabilities"],
    )

    txn = next(transactions, None)
    inventory = CounterInventory()

    prices = self.ledger.prices
    for date_range in filtered.interval_ranges(interval):
        while txn and txn.date < date_range.end:
            for posting in txn.postings:
                if posting.account.startswith(types):
                    inventory.add_position(posting)
            txn = next(transactions, None)
        yield DateAndBalance(
            date_range.end_inclusive,
            conv.apply(
                inventory,
                prices,
                date_range.end_inclusive,
            ),
        )

dumps(obj, **_kwargs)

Dump as a JSON string.

Source code in src/rustfava/core/charts.py

def dumps(obj: Any, **_kwargs: Any) -> str:
    """Dump as a JSON string."""
    return json.dumps(
        obj, sort_keys=True, separators=(",", ":"), default=_json_default
    )

loads(s)

Load a JSON string.

Source code in src/rustfava/core/charts.py

def loads(s: str | bytes) -> Any:
    """Load a JSON string."""
    return json.loads(s)

commodities

Attributes for auto-completion.

CommoditiesModule

Bases: FavaModule

Details about the currencies and commodities.

Source code in src/rustfava/core/commodities.py

class CommoditiesModule(FavaModule):
    """Details about the currencies and commodities."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        self.names: dict[str, str] = {}
        self.precisions: dict[str, int] = {}

    def load_file(self) -> None:  # noqa: D102
        self.names = {}
        self.precisions = {}
        for commodity in self.ledger.all_entries_by_type.Commodity:
            name = commodity.meta.get("name")
            if name:
                self.names[commodity.currency] = str(name)
            precision = commodity.meta.get("precision")
            if isinstance(precision, (str, int, Decimal)):
                with suppress(ValueError):
                    self.precisions[commodity.currency] = int(precision)

    def name(self, commodity: str) -> str:
        """Get the name of a commodity (or the commodity itself if not set)."""
        return self.names.get(commodity, commodity)

name(commodity)

Get the name of a commodity (or the commodity itself if not set).

Source code in src/rustfava/core/commodities.py

def name(self, commodity: str) -> str:
    """Get the name of a commodity (or the commodity itself if not set)."""
    return self.names.get(commodity, commodity)

conversion

Commodity conversion helpers for Fava.

All functions in this module will be automatically added as template filters.

Conversion

Bases: ABC

A conversion.

Source code in src/rustfava/core/conversion.py

class Conversion(ABC):
    """A conversion."""

    @abstractmethod
    def apply(
        self,
        inventory: CounterInventory,
        prices: RustfavaPriceMap,
        date: datetime.date | None = None,
    ) -> SimpleCounterInventory:
        """Apply the conversion to an inventory (CounterInventory)."""

apply(inventory, prices, date=None) abstractmethod

Apply the conversion to an inventory (CounterInventory).

Source code in src/rustfava/core/conversion.py

@abstractmethod
def apply(
    self,
    inventory: CounterInventory,
    prices: RustfavaPriceMap,
    date: datetime.date | None = None,
) -> SimpleCounterInventory:
    """Apply the conversion to an inventory (CounterInventory)."""

get_cost(pos)

Return the total cost of a Position.

Source code in src/rustfava/core/conversion.py

def get_cost(pos: Position) -> Amount:
    """Return the total cost of a Position."""
    cost_ = pos.cost
    return (
        _Amount(cost_.number * pos.units.number, cost_.currency)
        if cost_ is not None
        else pos.units
    )

get_market_value(pos, prices, date=None)

Get the market value of a Position.

This differs from the convert.get_value function in Beancount by returning the cost value if no price can be found.

Parameters:

Name	Type	Description	Default
pos	Position	A Position.	required
prices	RustfavaPriceMap	A rustfavaPriceMap	required
date	date | None	A datetime.date instance to evaluate the value at, or None.	None

Returns:

Type	Description
Amount	An Amount, with value converted or if the conversion failed just the
Amount	cost value (or the units if the position has no cost).

Source code in src/rustfava/core/conversion.py

def get_market_value(
    pos: Position,
    prices: RustfavaPriceMap,
    date: datetime.date | None = None,
) -> Amount:
    """Get the market value of a Position.

    This differs from the convert.get_value function in Beancount by returning
    the cost value if no price can be found.

    Args:
        pos: A Position.
        prices: A rustfavaPriceMap
        date: A datetime.date instance to evaluate the value at, or None.

    Returns:
        An Amount, with value converted or if the conversion failed just the
        cost value (or the units if the position has no cost).
    """
    units_ = pos.units
    cost_ = pos.cost

    if cost_ is not None:
        value_currency = cost_.currency
        base_quote = (units_.currency, value_currency)
        price_number = prices.get_price(base_quote, date)
        if price_number is not None:
            return _Amount(
                units_.number * price_number,
                value_currency,
            )
        return _Amount(units_.number * cost_.number, value_currency)
    return units_

convert_position(pos, target_currency, prices, date=None)

Get the value of a Position in a particular currency.

Parameters:

Name	Type	Description	Default
pos	Position	A Position.	required
target_currency	str	The target currency to convert to.	required
prices	RustfavaPriceMap	A rustfavaPriceMap	required
date	date | None	A datetime.date instance to evaluate the value at, or None.	None

Returns:

Type	Description
Amount	An Amount, with value converted or if the conversion failed just the
Amount	cost value (or the units if the position has no cost).

Source code in src/rustfava/core/conversion.py

def convert_position(
    pos: Position,
    target_currency: str,
    prices: RustfavaPriceMap,
    date: datetime.date | None = None,
) -> Amount:
    """Get the value of a Position in a particular currency.

    Args:
        pos: A Position.
        target_currency: The target currency to convert to.
        prices: A rustfavaPriceMap
        date: A datetime.date instance to evaluate the value at, or None.

    Returns:
        An Amount, with value converted or if the conversion failed just the
        cost value (or the units if the position has no cost).
    """
    units_ = pos.units

    # try the direct conversion
    base_quote = (units_.currency, target_currency)
    price_number = prices.get_price(base_quote, date)
    if price_number is not None:
        return _Amount(units_.number * price_number, target_currency)

    cost_ = pos.cost
    if cost_ is not None:
        cost_currency = cost_.currency
        if cost_currency != target_currency:
            base_quote1 = (units_.currency, cost_currency)
            rate1 = prices.get_price(base_quote1, date)
            if rate1 is not None:
                base_quote2 = (cost_currency, target_currency)
                rate2 = prices.get_price(base_quote2, date)
                if rate2 is not None:
                    return _Amount(
                        units_.number * rate1 * rate2,
                        target_currency,
                    )
    return units_

conversion_from_str(value)

Parse a conversion string.

Source code in src/rustfava/core/conversion.py

def conversion_from_str(value: str | Conversion) -> Conversion:
    """Parse a conversion string."""
    if not isinstance(value, str):
        return value
    if value == "at_cost":
        return AT_COST
    if value == "at_value":
        return AT_VALUE
    if value == "units":
        return UNITS

    return _CurrencyConversion(value)

cost_or_value(inventory, conversion, prices, date=None)

Get the cost or value of an inventory.

Source code in src/rustfava/core/conversion.py

def cost_or_value(
    inventory: CounterInventory,
    conversion: str | Conversion,
    prices: RustfavaPriceMap,
    date: datetime.date | None = None,
) -> SimpleCounterInventory:
    """Get the cost or value of an inventory."""
    conversion = conversion_from_str(conversion)
    return conversion.apply(inventory, prices, date)

documents

Document path related helpers.

NotADocumentsFolderError

Bases: RustfavaAPIError

Not a documents folder.

Source code in src/rustfava/core/documents.py

class NotADocumentsFolderError(RustfavaAPIError):
    """Not a documents folder."""

    def __init__(self, folder: str) -> None:
        super().__init__(f"Not a documents folder: {folder}.")

NotAValidAccountError

Bases: RustfavaAPIError

Not a valid account.

Source code in src/rustfava/core/documents.py

class NotAValidAccountError(RustfavaAPIError):
    """Not a valid account."""

    def __init__(self, account: str) -> None:
        super().__init__(f"Not a valid account: '{account}'")

is_document_or_import_file(filename, ledger)

Check whether the filename is a document or in an import directory.

This is a security validation function that prevents path traversal.

Parameters:

Name	Type	Description	Default
filename	str	The filename to check.	required
ledger	RustfavaLedger	The RustfavaLedger.	required

Returns:

Type	Description
bool	Whether this is one of the documents or a path in an import dir.

Source code in src/rustfava/core/documents.py

def is_document_or_import_file(filename: str, ledger: RustfavaLedger) -> bool:
    """Check whether the filename is a document or in an import directory.

    This is a security validation function that prevents path traversal.

    Args:
        filename: The filename to check.
        ledger: The RustfavaLedger.

    Returns:
        Whether this is one of the documents or a path in an import dir.
    """
    # Check if it's an exact match for a known document
    if any(
        filename == d.filename for d in ledger.all_entries_by_type.Document
    ):
        return True
    # Check if resolved path is within an import directory (prevents path traversal)
    file_path = Path(filename).resolve()
    for import_dir in ledger.fava_options.import_dirs:
        resolved_dir = ledger.join_path(import_dir).resolve()
        if file_path.is_relative_to(resolved_dir):
            return True
    return False

filepath_in_document_folder(documents_folder, account, filename, ledger)

File path for a document in the folder for an account.

Parameters:

Name	Type	Description	Default
documents_folder	str	The documents folder.	required
account	str	The account to choose the subfolder for.	required
filename	str	The filename of the document.	required
ledger	RustfavaLedger	The RustfavaLedger.	required

Returns:

Type	Description
Path	The path that the document should be saved at.

Source code in src/rustfava/core/documents.py

def filepath_in_document_folder(
    documents_folder: str,
    account: str,
    filename: str,
    ledger: RustfavaLedger,
) -> Path:
    """File path for a document in the folder for an account.

    Args:
        documents_folder: The documents folder.
        account: The account to choose the subfolder for.
        filename: The filename of the document.
        ledger: The RustfavaLedger.

    Returns:
        The path that the document should be saved at.
    """
    if documents_folder not in ledger.options["documents"]:
        raise NotADocumentsFolderError(documents_folder)

    if account not in ledger.attributes.accounts:
        raise NotAValidAccountError(account)

    filename = filename.replace(sep, " ")
    if altsep:  # pragma: no cover
        filename = filename.replace(altsep, " ")

    return ledger.join_path(
        documents_folder,
        *account.split(":"),
        filename,
    )

extensions

Fava extensions.

ExtensionDetails dataclass

The information about an extension that is needed for the frontend.

Source code in src/rustfava/core/extensions.py

@dataclass
class ExtensionDetails:
    """The information about an extension that is needed for the frontend."""

    name: str
    report_title: str | None
    has_js_module: bool

ExtensionModule

Bases: FavaModule

Fava extensions.

Source code in src/rustfava/core/extensions.py

class ExtensionModule(FavaModule):
    """Fava extensions."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        self._instances: dict[str, RustfavaExtensionBase] = {}
        self._loaded_extensions: set[type[RustfavaExtensionBase]] = set()
        self.errors: list[RustfavaExtensionError] = []

    def load_file(self) -> None:  # noqa: D102
        self.errors = []

        custom_entries = self.ledger.all_entries_by_type.Custom

        seen = set()
        for entry in (e for e in custom_entries if e.type == "fava-extension"):
            extension = entry.values[0].value
            if extension in seen:  # pragma: no cover
                self.errors.append(
                    RustfavaExtensionError(
                        entry.meta, f"Duplicate extension '{extension}'", entry
                    )
                )
                continue

            seen.add(extension)
            extensions, errors = find_extensions(
                Path(self.ledger.beancount_file_path).parent,
                extension,
            )
            self.errors.extend(errors)

            for cls in extensions:
                ext_config = (
                    entry.values[1].value if len(entry.values) > 1 else None
                )
                if cls not in self._loaded_extensions:
                    self._loaded_extensions.add(cls)
                    try:
                        ext = cls(self.ledger, ext_config)
                        self._instances[ext.name] = ext
                    except ExtensionConfigError as error:  # pragma: no cover
                        self.errors.append(
                            RustfavaExtensionError(entry.meta, str(error), entry)
                        )

    @property
    def _exts(self) -> Iterable[RustfavaExtensionBase]:
        return self._instances.values()

    @property
    def extension_details(self) -> Sequence[ExtensionDetails]:
        """Extension information to provide to the Frontend."""
        return [
            ExtensionDetails(ext.name, ext.report_title, ext.has_js_module)
            for ext in self._exts
        ]

    def get_extension(self, name: str) -> RustfavaExtensionBase | None:
        """Get the extension with the given name."""
        return self._instances.get(name, None)

    def after_load_file(self) -> None:
        """Run all `after_load_file` hooks."""
        for ext in self._exts:
            ext.after_load_file()

    def before_request(self) -> None:
        """Run all `before_request` hooks."""
        for ext in self._exts:
            ext.before_request()

    def after_entry_modified(self, entry: Directive, new_lines: str) -> None:
        """Run all `after_entry_modified` hooks."""
        for ext in self._exts:  # pragma: no cover
            ext.after_entry_modified(entry, new_lines)

    def after_insert_entry(self, entry: Directive) -> None:
        """Run all `after_insert_entry` hooks."""
        for ext in self._exts:  # pragma: no cover
            ext.after_insert_entry(entry)

    def after_delete_entry(self, entry: Directive) -> None:
        """Run all `after_delete_entry` hooks."""
        for ext in self._exts:  # pragma: no cover
            ext.after_delete_entry(entry)

    def after_insert_metadata(
        self,
        entry: Directive,
        key: str,
        value: str,
    ) -> None:
        """Run all `after_insert_metadata` hooks."""
        for ext in self._exts:  # pragma: no cover
            ext.after_insert_metadata(entry, key, value)

    def after_write_source(self, path: str, source: str) -> None:
        """Run all `after_write_source` hooks."""
        for ext in self._exts:
            ext.after_write_source(path, source)

extension_details property

Extension information to provide to the Frontend.

get_extension(name)

Get the extension with the given name.

Source code in src/rustfava/core/extensions.py

def get_extension(self, name: str) -> RustfavaExtensionBase | None:
    """Get the extension with the given name."""
    return self._instances.get(name, None)

after_load_file()

Run all after_load_file hooks.

Source code in src/rustfava/core/extensions.py

def after_load_file(self) -> None:
    """Run all `after_load_file` hooks."""
    for ext in self._exts:
        ext.after_load_file()

before_request()

Run all before_request hooks.

Source code in src/rustfava/core/extensions.py

def before_request(self) -> None:
    """Run all `before_request` hooks."""
    for ext in self._exts:
        ext.before_request()

after_entry_modified(entry, new_lines)

Run all after_entry_modified hooks.

Source code in src/rustfava/core/extensions.py

def after_entry_modified(self, entry: Directive, new_lines: str) -> None:
    """Run all `after_entry_modified` hooks."""
    for ext in self._exts:  # pragma: no cover
        ext.after_entry_modified(entry, new_lines)

after_insert_entry(entry)

Run all after_insert_entry hooks.

Source code in src/rustfava/core/extensions.py

def after_insert_entry(self, entry: Directive) -> None:
    """Run all `after_insert_entry` hooks."""
    for ext in self._exts:  # pragma: no cover
        ext.after_insert_entry(entry)

after_delete_entry(entry)

Run all after_delete_entry hooks.

Source code in src/rustfava/core/extensions.py

def after_delete_entry(self, entry: Directive) -> None:
    """Run all `after_delete_entry` hooks."""
    for ext in self._exts:  # pragma: no cover
        ext.after_delete_entry(entry)

after_insert_metadata(entry, key, value)

Run all after_insert_metadata hooks.

Source code in src/rustfava/core/extensions.py

def after_insert_metadata(
    self,
    entry: Directive,
    key: str,
    value: str,
) -> None:
    """Run all `after_insert_metadata` hooks."""
    for ext in self._exts:  # pragma: no cover
        ext.after_insert_metadata(entry, key, value)

after_write_source(path, source)

Run all after_write_source hooks.

Source code in src/rustfava/core/extensions.py

def after_write_source(self, path: str, source: str) -> None:
    """Run all `after_write_source` hooks."""
    for ext in self._exts:
        ext.after_write_source(path, source)

fava_options

rustfava's options.

Options for rustfava can be specified through Custom entries in the Beancount file. This module contains a list of possible options, the defaults and the code for parsing the options.

OptionError dataclass

Bases: BeancountError

An error for one of the rustfava options.

Source code in src/rustfava/core/fava_options.py

class OptionError(BeancountError):
    """An error for one of the rustfava options."""

InsertEntryOption dataclass

Insert option.

An option that determines where entries for matching accounts should be inserted.

Source code in src/rustfava/core/fava_options.py

@dataclass(frozen=True)
class InsertEntryOption:
    """Insert option.

    An option that determines where entries for matching accounts should be
    inserted.
    """

    date: datetime.date
    re: Pattern[str]
    filename: str
    lineno: int

RustfavaOptions dataclass

Options for rustfava that can be set in the Beancount file.

Source code in src/rustfava/core/fava_options.py

@dataclass
class RustfavaOptions:
    """Options for rustfava that can be set in the Beancount file."""

    account_journal_include_children: bool = True
    auto_reload: bool = False
    collapse_pattern: Sequence[Pattern[str]] = field(default_factory=list)
    conversion_currencies: tuple[str, ...] = ()
    currency_column: int = 61
    default_file: str | None = None
    default_page: str = "income_statement/"
    fiscal_year_end: FiscalYearEnd = END_OF_YEAR
    import_config: str | None = None
    import_dirs: Sequence[str] = field(default_factory=list)
    indent: int = 2
    insert_entry: Sequence[InsertEntryOption] = field(default_factory=list)
    invert_gains_losses_colors: bool = False
    invert_income_liabilities_equity: bool = False
    language: str | None = None
    locale: str | None = None
    show_accounts_with_zero_balance: bool = True
    show_accounts_with_zero_transactions: bool = True
    show_closed_accounts: bool = False
    sidebar_show_queries: int = 5
    unrealized: str = "Unrealized"
    upcoming_events: int = 7
    uptodate_indicator_grey_lookback_days: int = 60
    use_external_editor: bool = False

    def set_collapse_pattern(self, value: str) -> None:
        """Set the collapse_pattern option."""
        try:
            pattern = re.compile(value)
        except re.error as err:
            raise NotARegularExpressionError(value) from err
        # It's typed as Sequence so that it's not externally mutated
        self.collapse_pattern.append(pattern)  # type: ignore[attr-defined]

    def set_default_file(self, value: str, filename: str) -> None:
        """Set the default_file option."""
        self.default_file = (
            str((Path(filename).parent / value).absolute())
            if value
            else filename
        )

    def set_fiscal_year_end(self, value: str) -> None:
        """Set the fiscal_year_end option."""
        fye = parse_fye_string(value)
        if fye is None:
            raise InvalidFiscalYearEndOptionError(value)
        self.fiscal_year_end = fye

    def set_import_dirs(self, value: str) -> None:
        """Add an import directory."""
        # It's typed as Sequence so that it's not externally mutated
        self.import_dirs.append(value)  # type: ignore[attr-defined]

    def set_insert_entry(
        self, value: str, date: datetime.date, filename: str, lineno: int
    ) -> None:
        """Set the insert_entry option."""
        try:
            pattern = re.compile(value)
        except re.error as err:
            raise NotARegularExpressionError(value) from err
        opt = InsertEntryOption(date, pattern, filename, lineno)
        # It's typed as Sequence so that it's not externally mutated
        self.insert_entry.append(opt)  # type: ignore[attr-defined]

    def set_language(self, value: str) -> None:
        """Set the locale option."""
        try:
            locale = Locale.parse(value)
            if (
                not locale.language == "en"
                and get_translations(locale) is None
            ):
                raise UnsupportedLanguageOptionError(value)
            self.language = value
        except UnknownLocaleError as err:
            raise UnknownLocaleOptionError(value) from err

    def set_locale(self, value: str) -> None:
        """Set the locale option."""
        try:
            Locale.parse(value)
            self.locale = value
        except UnknownLocaleError as err:
            raise UnknownLocaleOptionError(value) from err

set_collapse_pattern(value)

Set the collapse_pattern option.

Source code in src/rustfava/core/fava_options.py

def set_collapse_pattern(self, value: str) -> None:
    """Set the collapse_pattern option."""
    try:
        pattern = re.compile(value)
    except re.error as err:
        raise NotARegularExpressionError(value) from err
    # It's typed as Sequence so that it's not externally mutated
    self.collapse_pattern.append(pattern)  # type: ignore[attr-defined]

set_default_file(value, filename)

Set the default_file option.

Source code in src/rustfava/core/fava_options.py

def set_default_file(self, value: str, filename: str) -> None:
    """Set the default_file option."""
    self.default_file = (
        str((Path(filename).parent / value).absolute())
        if value
        else filename
    )

set_fiscal_year_end(value)

Set the fiscal_year_end option.

Source code in src/rustfava/core/fava_options.py

def set_fiscal_year_end(self, value: str) -> None:
    """Set the fiscal_year_end option."""
    fye = parse_fye_string(value)
    if fye is None:
        raise InvalidFiscalYearEndOptionError(value)
    self.fiscal_year_end = fye

set_import_dirs(value)

Add an import directory.

Source code in src/rustfava/core/fava_options.py

def set_import_dirs(self, value: str) -> None:
    """Add an import directory."""
    # It's typed as Sequence so that it's not externally mutated
    self.import_dirs.append(value)  # type: ignore[attr-defined]

set_insert_entry(value, date, filename, lineno)

Set the insert_entry option.

Source code in src/rustfava/core/fava_options.py

def set_insert_entry(
    self, value: str, date: datetime.date, filename: str, lineno: int
) -> None:
    """Set the insert_entry option."""
    try:
        pattern = re.compile(value)
    except re.error as err:
        raise NotARegularExpressionError(value) from err
    opt = InsertEntryOption(date, pattern, filename, lineno)
    # It's typed as Sequence so that it's not externally mutated
    self.insert_entry.append(opt)  # type: ignore[attr-defined]

set_language(value)

Set the locale option.

Source code in src/rustfava/core/fava_options.py

def set_language(self, value: str) -> None:
    """Set the locale option."""
    try:
        locale = Locale.parse(value)
        if (
            not locale.language == "en"
            and get_translations(locale) is None
        ):
            raise UnsupportedLanguageOptionError(value)
        self.language = value
    except UnknownLocaleError as err:
        raise UnknownLocaleOptionError(value) from err

set_locale(value)

Set the locale option.

Source code in src/rustfava/core/fava_options.py

def set_locale(self, value: str) -> None:
    """Set the locale option."""
    try:
        Locale.parse(value)
        self.locale = value
    except UnknownLocaleError as err:
        raise UnknownLocaleOptionError(value) from err

parse_option_custom_entry(entry, options)

Parse a single custom fava-option entry and set option accordingly.

Source code in src/rustfava/core/fava_options.py

def parse_option_custom_entry(  # noqa: PLR0912
    entry: Custom,
    options: RustfavaOptions,
) -> None:
    """Parse a single custom fava-option entry and set option accordingly."""
    key = str(entry.values[0].value).replace("-", "_")
    if key not in All_OPTS:
        raise UnknownOptionError(key)

    value = entry.values[1].value if len(entry.values) > 1 else ""
    if not isinstance(value, str):
        raise NotAStringOptionError(key)
    filename, lineno = get_position(entry)

    if key == "collapse_pattern":
        options.set_collapse_pattern(value)
    elif key == "default_file":
        options.set_default_file(value, filename)
    elif key == "fiscal_year_end":
        options.set_fiscal_year_end(value)
    elif key == "import_dirs":
        options.set_import_dirs(value)
    elif key == "insert_entry":
        options.set_insert_entry(value, entry.date, filename, lineno)
    elif key == "language":
        options.set_language(value)
    elif key == "locale":
        options.set_locale(value)
    elif key in STR_OPTS:
        setattr(options, key, value)
    elif key in BOOL_OPTS:
        setattr(options, key, value.lower() == "true")
    elif key in INT_OPTS:
        setattr(options, key, int(value))
    else:  # key in TUPLE_OPTS
        setattr(options, key, tuple(value.strip().split(" ")))

parse_options(custom_entries)

Parse custom entries for rustfava options.

The format for option entries is the following::

2016-04-01 custom "fava-option" "[name]" "[value]"

Parameters:

Name	Type	Description	Default
custom_entries	Sequence[Custom]	A list of Custom entries.	required

Returns:

Type	Description
RustfavaOptions	A tuple (options, errors) where options is a dictionary of all options
list[OptionError]	to values, and errors contains possible parsing errors.

Source code in src/rustfava/core/fava_options.py

def parse_options(
    custom_entries: Sequence[Custom],
) -> tuple[RustfavaOptions, list[OptionError]]:
    """Parse custom entries for rustfava options.

    The format for option entries is the following::

        2016-04-01 custom "fava-option" "[name]" "[value]"

    Args:
        custom_entries: A list of Custom entries.

    Returns:
        A tuple (options, errors) where options is a dictionary of all options
        to values, and errors contains possible parsing errors.
    """
    options = RustfavaOptions()
    errors = []

    for entry in (e for e in custom_entries if e.type == "fava-option"):
        try:
            if not entry.values:
                raise MissingOptionError
            parse_option_custom_entry(entry, options)
        except (IndexError, TypeError, ValueError) as err:
            msg = f"Failed to parse fava-option entry: {err!s}"
            errors.append(OptionError(entry.meta, msg, entry))

    return options, errors

file

Reading/writing Beancount files.

NonSourceFileError

Bases: RustfavaAPIError

Trying to read a non-source file.

Source code in src/rustfava/core/file.py

class NonSourceFileError(RustfavaAPIError):
    """Trying to read a non-source file."""

    def __init__(self, path: Path) -> None:
        super().__init__(f"Trying to read a non-source file at '{path}'")

ExternallyChangedError

Bases: RustfavaAPIError

The file changed externally.

Source code in src/rustfava/core/file.py

class ExternallyChangedError(RustfavaAPIError):
    """The file changed externally."""

    def __init__(self, path: Path) -> None:
        super().__init__(f"The file at '{path}' changed externally.")

GeneratedEntryError

Bases: RustfavaAPIError

The entry is generated and cannot be edited.

Source code in src/rustfava/core/file.py

class GeneratedEntryError(RustfavaAPIError):
    """The entry is generated and cannot be edited."""

    def __init__(self) -> None:
        super().__init__("The entry is generated and cannot be edited.")

InvalidUnicodeError

Bases: RustfavaAPIError

The source file contains invalid unicode.

Source code in src/rustfava/core/file.py

class InvalidUnicodeError(RustfavaAPIError):
    """The source file contains invalid unicode."""

    def __init__(self, reason: str) -> None:
        super().__init__(
            f"The source file contains invalid unicode: {reason}.",
        )

FileModule

Bases: FavaModule

Functions related to reading/writing to Beancount files.

Source code in src/rustfava/core/file.py

class FileModule(FavaModule):
    """Functions related to reading/writing to Beancount files."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        self._lock = threading.Lock()

    def get_source(self, path: Path) -> tuple[str, str]:
        """Get source files.

        Args:
            path: The path of the file.

        Returns:
            A string with the file contents and the `sha256sum` of the file.

        Raises:
            NonSourceFileError: If the file is not one of the source files.
            InvalidUnicodeError: If the file contains invalid unicode.
        """
        if str(path) not in self.ledger.options["include"]:
            raise NonSourceFileError(path)

        try:
            source = path.read_text("utf-8")
        except UnicodeDecodeError as exc:
            raise InvalidUnicodeError(str(exc)) from exc

        return source, _sha256_str(source)

    def set_source(self, path: Path, source: str, sha256sum: str) -> str:
        """Write to source file.

        Args:
            path: The path of the file.
            source: A string with the file contents.
            sha256sum: Hash of the file.

        Returns:
            The `sha256sum` of the updated file.

        Raises:
            NonSourceFileError: If the file is not one of the source files.
            InvalidUnicodeError: If the file contains invalid unicode.
            ExternallyChangedError: If the file was changed externally.
        """
        with self._lock:
            _, original_sha256sum = self.get_source(path)
            if original_sha256sum != sha256sum:
                raise ExternallyChangedError(path)

            newline = _file_newline_character(path)
            with path.open("w", encoding="utf-8", newline=newline) as file:
                file.write(source)
            self.ledger.watcher.notify(path)

            self.ledger.extensions.after_write_source(str(path), source)
            self.ledger.load_file()

            return _sha256_str(source)

    def insert_metadata(
        self,
        entry_hash: str,
        basekey: str,
        value: str,
    ) -> None:
        """Insert metadata into a file at lineno.

        Also, prevent duplicate keys.

        Args:
            entry_hash: Hash of an entry.
            basekey: Key to insert metadata for.
            value: Metadate value to insert.
        """
        with self._lock:
            self.ledger.changed()
            entry = self.ledger.get_entry(entry_hash)
            key = next_key(basekey, entry.meta)
            indent = self.ledger.fava_options.indent
            path, lineno = _get_position(entry)
            insert_metadata_in_file(path, lineno, indent, key, value)
            self.ledger.watcher.notify(path)
            self.ledger.extensions.after_insert_metadata(entry, key, value)

    def save_entry_slice(
        self,
        entry_hash: str,
        source_slice: str,
        sha256sum: str,
    ) -> str:
        """Save slice of the source file for an entry.

        Args:
            entry_hash: Hash of an entry.
            source_slice: The lines that the entry should be replaced with.
            sha256sum: The sha256sum of the current lines of the entry.

        Returns:
            The `sha256sum` of the new lines of the entry.

        Raises:
            RustfavaAPIError: If the entry is not found or the file changed.
        """
        with self._lock:
            entry = self.ledger.get_entry(entry_hash)
            new_sha256sum = save_entry_slice(entry, source_slice, sha256sum)
            self.ledger.watcher.notify(Path(get_position(entry)[0]))
            self.ledger.extensions.after_entry_modified(entry, source_slice)
            return new_sha256sum

    def delete_entry_slice(self, entry_hash: str, sha256sum: str) -> None:
        """Delete slice of the source file for an entry.

        Args:
            entry_hash: Hash of an entry.
            sha256sum: The sha256sum of the current lines of the entry.

        Raises:
            RustfavaAPIError: If the entry is not found or the file changed.
        """
        with self._lock:
            entry = self.ledger.get_entry(entry_hash)
            delete_entry_slice(entry, sha256sum)
            self.ledger.watcher.notify(Path(get_position(entry)[0]))
            self.ledger.extensions.after_delete_entry(entry)

    def insert_entries(self, entries: Sequence[Directive]) -> None:
        """Insert entries.

        Args:
            entries: A list of entries.
        """
        with self._lock:
            self.ledger.changed()
            fava_options = self.ledger.fava_options
            for entry in sorted(entries, key=_incomplete_sortkey):
                path, updated_insert_options = insert_entry(
                    entry,
                    (
                        self.ledger.fava_options.default_file
                        or self.ledger.beancount_file_path
                    ),
                    insert_options=fava_options.insert_entry,
                    currency_column=fava_options.currency_column,
                    indent=fava_options.indent,
                )
                self.ledger.watcher.notify(path)
                self.ledger.fava_options.insert_entry = updated_insert_options
                self.ledger.extensions.after_insert_entry(entry)

    def render_entries(self, entries: Sequence[Directive]) -> Iterable[Markup]:
        """Return entries in Beancount format.

        Only renders :class:`.Balance` and :class:`.Transaction`.

        Args:
            entries: A list of entries.

        Yields:
            The entries rendered in Beancount format.
        """
        indent = self.ledger.fava_options.indent
        for entry in entries:
            if isinstance(entry, (Balance, Transaction)):
                if (
                    isinstance(entry, Transaction)
                    and entry.flag in _EXCL_FLAGS
                ):
                    continue
                try:
                    yield Markup(get_entry_slice(entry)[0] + "\n")  # noqa: S704
                except (KeyError, FileNotFoundError):
                    yield Markup(  # noqa: S704
                        to_string(
                            entry,
                            self.ledger.fava_options.currency_column,
                            indent,
                        ),
                    )

get_source(path)

Get source files.

Parameters:

Name	Type	Description	Default
path	Path	The path of the file.	required

Returns:

Type	Description
tuple[str, str]	A string with the file contents and the sha256sum of the file.

Raises:

Type	Description
NonSourceFileError	If the file is not one of the source files.
InvalidUnicodeError	If the file contains invalid unicode.

Source code in src/rustfava/core/file.py

def get_source(self, path: Path) -> tuple[str, str]:
    """Get source files.

    Args:
        path: The path of the file.

    Returns:
        A string with the file contents and the `sha256sum` of the file.

    Raises:
        NonSourceFileError: If the file is not one of the source files.
        InvalidUnicodeError: If the file contains invalid unicode.
    """
    if str(path) not in self.ledger.options["include"]:
        raise NonSourceFileError(path)

    try:
        source = path.read_text("utf-8")
    except UnicodeDecodeError as exc:
        raise InvalidUnicodeError(str(exc)) from exc

    return source, _sha256_str(source)

set_source(path, source, sha256sum)

Write to source file.

Parameters:

Name	Type	Description	Default
path	Path	The path of the file.	required
source	str	A string with the file contents.	required
sha256sum	str	Hash of the file.	required

Returns:

Type	Description
str	The sha256sum of the updated file.

Raises:

Type	Description
NonSourceFileError	If the file is not one of the source files.
InvalidUnicodeError	If the file contains invalid unicode.
ExternallyChangedError	If the file was changed externally.

Source code in src/rustfava/core/file.py

def set_source(self, path: Path, source: str, sha256sum: str) -> str:
    """Write to source file.

    Args:
        path: The path of the file.
        source: A string with the file contents.
        sha256sum: Hash of the file.

    Returns:
        The `sha256sum` of the updated file.

    Raises:
        NonSourceFileError: If the file is not one of the source files.
        InvalidUnicodeError: If the file contains invalid unicode.
        ExternallyChangedError: If the file was changed externally.
    """
    with self._lock:
        _, original_sha256sum = self.get_source(path)
        if original_sha256sum != sha256sum:
            raise ExternallyChangedError(path)

        newline = _file_newline_character(path)
        with path.open("w", encoding="utf-8", newline=newline) as file:
            file.write(source)
        self.ledger.watcher.notify(path)

        self.ledger.extensions.after_write_source(str(path), source)
        self.ledger.load_file()

        return _sha256_str(source)

insert_metadata(entry_hash, basekey, value)

Insert metadata into a file at lineno.

Also, prevent duplicate keys.

Parameters:

Name	Type	Description	Default
entry_hash	str	Hash of an entry.	required
basekey	str	Key to insert metadata for.	required
value	str	Metadate value to insert.	required

Source code in src/rustfava/core/file.py

def insert_metadata(
    self,
    entry_hash: str,
    basekey: str,
    value: str,
) -> None:
    """Insert metadata into a file at lineno.

    Also, prevent duplicate keys.

    Args:
        entry_hash: Hash of an entry.
        basekey: Key to insert metadata for.
        value: Metadate value to insert.
    """
    with self._lock:
        self.ledger.changed()
        entry = self.ledger.get_entry(entry_hash)
        key = next_key(basekey, entry.meta)
        indent = self.ledger.fava_options.indent
        path, lineno = _get_position(entry)
        insert_metadata_in_file(path, lineno, indent, key, value)
        self.ledger.watcher.notify(path)
        self.ledger.extensions.after_insert_metadata(entry, key, value)

save_entry_slice(entry_hash, source_slice, sha256sum)

Save slice of the source file for an entry.

Parameters:

Name	Type	Description	Default
entry_hash	str	Hash of an entry.	required
source_slice	str	The lines that the entry should be replaced with.	required
sha256sum	str	The sha256sum of the current lines of the entry.	required

Returns:

Type	Description
str	The sha256sum of the new lines of the entry.

Raises:

Type	Description
RustfavaAPIError	If the entry is not found or the file changed.

Source code in src/rustfava/core/file.py

def save_entry_slice(
    self,
    entry_hash: str,
    source_slice: str,
    sha256sum: str,
) -> str:
    """Save slice of the source file for an entry.

    Args:
        entry_hash: Hash of an entry.
        source_slice: The lines that the entry should be replaced with.
        sha256sum: The sha256sum of the current lines of the entry.

    Returns:
        The `sha256sum` of the new lines of the entry.

    Raises:
        RustfavaAPIError: If the entry is not found or the file changed.
    """
    with self._lock:
        entry = self.ledger.get_entry(entry_hash)
        new_sha256sum = save_entry_slice(entry, source_slice, sha256sum)
        self.ledger.watcher.notify(Path(get_position(entry)[0]))
        self.ledger.extensions.after_entry_modified(entry, source_slice)
        return new_sha256sum

delete_entry_slice(entry_hash, sha256sum)

Delete slice of the source file for an entry.

Parameters:

Name	Type	Description	Default
entry_hash	str	Hash of an entry.	required
sha256sum	str	The sha256sum of the current lines of the entry.	required

Raises:

Type	Description
RustfavaAPIError	If the entry is not found or the file changed.

Source code in src/rustfava/core/file.py

def delete_entry_slice(self, entry_hash: str, sha256sum: str) -> None:
    """Delete slice of the source file for an entry.

    Args:
        entry_hash: Hash of an entry.
        sha256sum: The sha256sum of the current lines of the entry.

    Raises:
        RustfavaAPIError: If the entry is not found or the file changed.
    """
    with self._lock:
        entry = self.ledger.get_entry(entry_hash)
        delete_entry_slice(entry, sha256sum)
        self.ledger.watcher.notify(Path(get_position(entry)[0]))
        self.ledger.extensions.after_delete_entry(entry)

insert_entries(entries)

Insert entries.

Parameters:

Name	Type	Description	Default
entries	Sequence[Directive]	A list of entries.	required

Source code in src/rustfava/core/file.py

def insert_entries(self, entries: Sequence[Directive]) -> None:
    """Insert entries.

    Args:
        entries: A list of entries.
    """
    with self._lock:
        self.ledger.changed()
        fava_options = self.ledger.fava_options
        for entry in sorted(entries, key=_incomplete_sortkey):
            path, updated_insert_options = insert_entry(
                entry,
                (
                    self.ledger.fava_options.default_file
                    or self.ledger.beancount_file_path
                ),
                insert_options=fava_options.insert_entry,
                currency_column=fava_options.currency_column,
                indent=fava_options.indent,
            )
            self.ledger.watcher.notify(path)
            self.ledger.fava_options.insert_entry = updated_insert_options
            self.ledger.extensions.after_insert_entry(entry)

render_entries(entries)

Return entries in Beancount format.

Only renders :class:.Balance and :class:.Transaction.

Parameters:

Name	Type	Description	Default
entries	Sequence[Directive]	A list of entries.	required

Yields:

Type	Description
Iterable[Markup]	The entries rendered in Beancount format.

Source code in src/rustfava/core/file.py

def render_entries(self, entries: Sequence[Directive]) -> Iterable[Markup]:
    """Return entries in Beancount format.

    Only renders :class:`.Balance` and :class:`.Transaction`.

    Args:
        entries: A list of entries.

    Yields:
        The entries rendered in Beancount format.
    """
    indent = self.ledger.fava_options.indent
    for entry in entries:
        if isinstance(entry, (Balance, Transaction)):
            if (
                isinstance(entry, Transaction)
                and entry.flag in _EXCL_FLAGS
            ):
                continue
            try:
                yield Markup(get_entry_slice(entry)[0] + "\n")  # noqa: S704
            except (KeyError, FileNotFoundError):
                yield Markup(  # noqa: S704
                    to_string(
                        entry,
                        self.ledger.fava_options.currency_column,
                        indent,
                    ),
                )

insert_metadata_in_file(path, lineno, indent, key, value)

Insert the specified metadata in the file below lineno.

Takes the whitespace in front of the line that lineno into account.

Source code in src/rustfava/core/file.py

def insert_metadata_in_file(
    path: Path,
    lineno: int,
    indent: int,
    key: str,
    value: str,
) -> None:
    """Insert the specified metadata in the file below lineno.

    Takes the whitespace in front of the line that lineno into account.
    """
    with path.open(encoding="utf-8") as file:
        contents = file.readlines()

    contents.insert(lineno, f'{" " * indent}{key}: "{value}"\n')
    newline = _file_newline_character(path)
    with path.open("w", encoding="utf-8", newline=newline) as file:
        file.write("".join(contents))

find_entry_lines(lines, lineno)

Lines of entry starting at lineno.

Parameters:

Name	Type	Description	Default
lines	Sequence[str]	A list of lines.	required
lineno	int	The 0-based line-index to start at.	required

Source code in src/rustfava/core/file.py

def find_entry_lines(lines: Sequence[str], lineno: int) -> Sequence[str]:
    """Lines of entry starting at lineno.

    Args:
        lines: A list of lines.
        lineno: The 0-based line-index to start at.
    """
    entry_lines = [lines[lineno]]
    while True:
        lineno += 1
        try:
            line = lines[lineno]
        except IndexError:
            return entry_lines
        if not line.strip() or re.match(r"\S", line[0]):
            return entry_lines
        entry_lines.append(line)

get_entry_slice(entry)

Get slice of the source file for an entry.

Parameters:

Name	Type	Description	Default
entry	Directive	An entry.	required

Returns:

Type	Description
str	A string containing the lines of the entry and the sha256sum of
str	these lines.

Raises:

Type	Description
GeneratedEntryError	If the entry is generated and cannot be edited.

Source code in src/rustfava/core/file.py

def get_entry_slice(entry: Directive) -> tuple[str, str]:
    """Get slice of the source file for an entry.

    Args:
        entry: An entry.

    Returns:
        A string containing the lines of the entry and the `sha256sum` of
        these lines.

    Raises:
        GeneratedEntryError: If the entry is generated and cannot be edited.
    """
    path, lineno = _get_position(entry)
    with path.open(encoding="utf-8") as file:
        lines = file.readlines()

    entry_lines = find_entry_lines(lines, lineno - 1)
    entry_source = "".join(entry_lines).rstrip("\n")

    return entry_source, _sha256_str(entry_source)

save_entry_slice(entry, source_slice, sha256sum)

Save slice of the source file for an entry.

Parameters:

Name	Type	Description	Default
entry	Directive	An entry.	required
source_slice	str	The lines that the entry should be replaced with.	required
sha256sum	str	The sha256sum of the current lines of the entry.	required

Returns:

Type	Description
str	The sha256sum of the new lines of the entry.

Raises:

Type	Description
ExternallyChangedError	If the file was changed externally.
GeneratedEntryError	If the entry is generated and cannot be edited.

Source code in src/rustfava/core/file.py

def save_entry_slice(
    entry: Directive,
    source_slice: str,
    sha256sum: str,
) -> str:
    """Save slice of the source file for an entry.

    Args:
        entry: An entry.
        source_slice: The lines that the entry should be replaced with.
        sha256sum: The sha256sum of the current lines of the entry.

    Returns:
        The `sha256sum` of the new lines of the entry.

    Raises:
        ExternallyChangedError: If the file was changed externally.
        GeneratedEntryError: If the entry is generated and cannot be edited.
    """
    path, lineno = _get_position(entry)
    with path.open(encoding="utf-8") as file:
        lines = file.readlines()

    first_entry_line = lineno - 1
    entry_lines = find_entry_lines(lines, first_entry_line)
    entry_source = "".join(entry_lines).rstrip("\n")
    if _sha256_str(entry_source) != sha256sum:
        raise ExternallyChangedError(path)

    lines = [
        *lines[:first_entry_line],
        source_slice + "\n",
        *lines[first_entry_line + len(entry_lines) :],
    ]
    newline = _file_newline_character(path)
    with path.open("w", encoding="utf-8", newline=newline) as file:
        file.writelines(lines)

    return _sha256_str(source_slice)

delete_entry_slice(entry, sha256sum)

Delete slice of the source file for an entry.

Parameters:

Name	Type	Description	Default
entry	Directive	An entry.	required
sha256sum	str	The sha256sum of the current lines of the entry.	required

Raises:

Type	Description
ExternallyChangedError	If the file was changed externally.
GeneratedEntryError	If the entry is generated and cannot be edited.

Source code in src/rustfava/core/file.py

def delete_entry_slice(
    entry: Directive,
    sha256sum: str,
) -> None:
    """Delete slice of the source file for an entry.

    Args:
        entry: An entry.
        sha256sum: The sha256sum of the current lines of the entry.

    Raises:
        ExternallyChangedError: If the file was changed externally.
        GeneratedEntryError: If the entry is generated and cannot be edited.
    """
    path, lineno = _get_position(entry)
    with path.open(encoding="utf-8") as file:
        lines = file.readlines()

    first_entry_line = lineno - 1
    entry_lines = find_entry_lines(lines, first_entry_line)
    entry_source = "".join(entry_lines).rstrip("\n")
    if _sha256_str(entry_source) != sha256sum:
        raise ExternallyChangedError(path)

    # Also delete the whitespace following this entry
    last_entry_line = first_entry_line + len(entry_lines)
    while True:
        try:
            line = lines[last_entry_line]
        except IndexError:
            break
        if line.strip():  # pragma: no cover
            break
        last_entry_line += 1  # pragma: no cover
    lines = lines[:first_entry_line] + lines[last_entry_line:]
    newline = _file_newline_character(path)
    with path.open("w", encoding="utf-8", newline=newline) as file:
        file.writelines(lines)

insert_entry(entry, default_filename, insert_options, currency_column, indent)

Insert an entry.

Parameters:

Name	Type	Description	Default
entry	Directive	An entry.	required
default_filename	str	The default file to insert into if no option matches.	required
insert_options	Sequence[InsertEntryOption]	Insert options.	required
currency_column	int	The column to align currencies at.	required
indent	int	Number of indent spaces.	required

Returns:

Type	Description
tuple[Path, Sequence[InsertEntryOption]]	A changed path and list of updated insert options.

Source code in src/rustfava/core/file.py

def insert_entry(
    entry: Directive,
    default_filename: str,
    insert_options: Sequence[InsertEntryOption],
    currency_column: int,
    indent: int,
) -> tuple[Path, Sequence[InsertEntryOption]]:
    """Insert an entry.

    Args:
        entry: An entry.
        default_filename: The default file to insert into if no option matches.
        insert_options: Insert options.
        currency_column: The column to align currencies at.
        indent: Number of indent spaces.

    Returns:
        A changed path and list of updated insert options.
    """
    filename, lineno = find_insert_position(
        entry,
        insert_options,
        default_filename,
    )
    content = to_string(entry, currency_column, indent)

    path = Path(filename)
    with path.open(encoding="utf-8") as file:
        contents = file.readlines()

    if lineno is None:
        # Appending
        contents += "\n" + content
    else:
        contents.insert(lineno, content + "\n")

    newline = _file_newline_character(path)
    with path.open("w", encoding="utf-8", newline=newline) as file:
        file.writelines(contents)

    if lineno is None:
        return (path, insert_options)

    added_lines = content.count("\n") + 1
    return (
        path,
        [
            (
                replace(option, lineno=option.lineno + added_lines)
                if option.filename == filename and option.lineno > lineno
                else option
            )
            for option in insert_options
        ],
    )

find_insert_position(entry, insert_options, default_filename)

Find insert position for an entry.

Parameters:

Name	Type	Description	Default
entry	Directive	An entry.	required
insert_options	Sequence[InsertEntryOption]	A list of InsertOption.	required
default_filename	str	The default file to insert into if no option matches.	required

Returns:

Type	Description
tuple[str, int | None]	A tuple of the filename and the line number.

Source code in src/rustfava/core/file.py

def find_insert_position(
    entry: Directive,
    insert_options: Sequence[InsertEntryOption],
    default_filename: str,
) -> tuple[str, int | None]:
    """Find insert position for an entry.

    Args:
        entry: An entry.
        insert_options: A list of InsertOption.
        default_filename: The default file to insert into if no option matches.

    Returns:
        A tuple of the filename and the line number.
    """
    # Get the list of accounts that should be considered for the entry.
    # For transactions, we want the reversed list of posting accounts.
    accounts = get_entry_accounts(entry)

    # Make no assumptions about the order of insert_options entries and instead
    # sort them ourselves (by descending dates)
    insert_options = sorted(
        insert_options,
        key=attrgetter("date"),
        reverse=True,
    )

    for account in accounts:
        for insert_option in insert_options:
            # Only consider InsertOptions before the entry date.
            if insert_option.date >= entry.date:
                continue
            if insert_option.re.match(account):
                return (insert_option.filename, insert_option.lineno - 1)

    return (default_filename, None)

filters

Entry filters.

FilterError

Bases: RustfavaAPIError

Filter exception.

Source code in src/rustfava/core/filters.py

class FilterError(RustfavaAPIError):
    """Filter exception."""

    def __init__(self, filter_type: str, message: str) -> None:
        super().__init__(message)
        self.filter_type = filter_type

    def __str__(self) -> str:
        return self.message

FilterParseError

Bases: FilterError

Filter parse error.

Source code in src/rustfava/core/filters.py

class FilterParseError(FilterError):
    """Filter parse error."""

    def __init__(self) -> None:
        super().__init__("filter", "Failed to parse filter: ")

FilterIllegalCharError

Bases: FilterError

Filter illegal char error.

Source code in src/rustfava/core/filters.py

class FilterIllegalCharError(FilterError):
    """Filter illegal char error."""

    def __init__(self, char: str) -> None:
        super().__init__(
            "filter",
            f'Illegal character "{char}" in filter.',
        )

TimeFilterParseError

Bases: FilterError

Time filter parse error.

Source code in src/rustfava/core/filters.py

class TimeFilterParseError(FilterError):
    """Time filter parse error."""

    def __init__(self, value: str) -> None:
        super().__init__("time", f"Failed to parse date: {value}")

Token

A token having a certain type and value.

The lexer attribute only exists since PLY writes to it in case of a parser error.

Source code in src/rustfava/core/filters.py

class Token:
    """A token having a certain type and value.

    The lexer attribute only exists since PLY writes to it in case of a parser
    error.
    """

    __slots__ = ("lexer", "type", "value")

    def __init__(self, type_: str, value: str) -> None:
        self.type = type_
        self.value = value

    def __repr__(self) -> str:  # pragma: no cover
        return f"Token({self.type}, {self.value})"

FilterSyntaxLexer

Lexer for Fava's filter syntax.

Source code in src/rustfava/core/filters.py

class FilterSyntaxLexer:
    """Lexer for Fava's filter syntax."""

    tokens = (
        "ANY",
        "ALL",
        "CMP_OP",
        "EQ_OP",
        "KEY",
        "LINK",
        "NUMBER",
        "STRING",
        "TAG",
    )

    RULES = (
        ("LINK", r"\^[A-Za-z0-9\-_/.]+"),
        ("TAG", r"\#[A-Za-z0-9\-_/.]+"),
        ("ALL", r"all\("),
        ("ANY", r"any\("),
        ("KEY", r"[a-z][a-zA-Z0-9\-_]+(?=\s*(:|=|>=|<=|<|>))"),
        ("EQ_OP", r":"),
        ("CMP_OP", r"(=|>=|<=|<|>)"),
        ("NUMBER", r"\d*\.?\d+"),
        ("STRING", r"""\w[-\w]*|"[^"]*"|'[^']*'"""),
    )

    regex = re.compile(
        "|".join((f"(?P<{name}>{rule})" for name, rule in RULES)),
    )

    def LINK(self, token: str, value: str) -> tuple[str, str]:  # noqa: N802
        return token, value[1:]

    def TAG(self, token: str, value: str) -> tuple[str, str]:  # noqa: N802
        return token, value[1:]

    def KEY(self, token: str, value: str) -> tuple[str, str]:  # noqa: N802
        return token, value

    def ALL(self, token: str, _: str) -> tuple[str, str]:  # noqa: N802
        return token, token

    def ANY(self, token: str, _: str) -> tuple[str, str]:  # noqa: N802
        return token, token

    def EQ_OP(self, token: str, value: str) -> tuple[str, str]:  # noqa: N802
        return token, value

    def CMP_OP(self, token: str, value: str) -> tuple[str, str]:  # noqa: N802
        return token, value

    def NUMBER(self, token: str, value: str) -> tuple[str, Decimal]:  # noqa: N802
        return token, Decimal(value)

    def STRING(self, token: str, value: str) -> tuple[str, str]:  # noqa: N802
        if value[0] in {'"', "'"}:
            return token, value[1:-1]
        return token, value

    def lex(self, data: str) -> Iterable[Token]:
        """A generator yielding all tokens in a given line.

        Arguments:
            data: A string, the line to lex.

        Yields:
            All Tokens in the line.
        """
        ignore = " \t"
        literals = "-,()"
        regex = self.regex.match

        pos = 0
        length = len(data)
        while pos < length:
            char = data[pos]
            if char in ignore:
                pos += 1
                continue
            match = regex(data, pos)
            if match:
                value = match.group()
                pos += len(value)
                token = match.lastgroup
                if token is None:  # pragma: no cover
                    msg = "Internal Error"
                    raise ValueError(msg)
                func: Callable[[str, str], tuple[str, str]] = getattr(
                    self,
                    token,
                )
                ret = func(token, value)
                yield Token(*ret)
            elif char in literals:
                yield Token(char, char)
                pos += 1
            else:
                raise FilterIllegalCharError(char)

lex(data)

A generator yielding all tokens in a given line.

Parameters:

Name	Type	Description	Default
data	str	A string, the line to lex.	required

Yields:

Type	Description
Iterable[Token]	All Tokens in the line.

Source code in src/rustfava/core/filters.py

def lex(self, data: str) -> Iterable[Token]:
    """A generator yielding all tokens in a given line.

    Arguments:
        data: A string, the line to lex.

    Yields:
        All Tokens in the line.
    """
    ignore = " \t"
    literals = "-,()"
    regex = self.regex.match

    pos = 0
    length = len(data)
    while pos < length:
        char = data[pos]
        if char in ignore:
            pos += 1
            continue
        match = regex(data, pos)
        if match:
            value = match.group()
            pos += len(value)
            token = match.lastgroup
            if token is None:  # pragma: no cover
                msg = "Internal Error"
                raise ValueError(msg)
            func: Callable[[str, str], tuple[str, str]] = getattr(
                self,
                token,
            )
            ret = func(token, value)
            yield Token(*ret)
        elif char in literals:
            yield Token(char, char)
            pos += 1
        else:
            raise FilterIllegalCharError(char)

Match

Match a string.

Source code in src/rustfava/core/filters.py

class Match:
    """Match a string."""

    __slots__ = ("match",)

    match: Callable[[str], bool]

    def __init__(self, search: str) -> None:
        try:
            match = re.compile(search, re.IGNORECASE).search
            self.match = lambda s: bool(match(s))
        except re.error:
            self.match = lambda s: s == search

    def __call__(self, obj: Any) -> bool:
        return self.match(str(obj))

MatchAmount

Matches an amount.

Source code in src/rustfava/core/filters.py

class MatchAmount:
    """Matches an amount."""

    __slots__ = ("match",)

    match: Callable[[Decimal], bool]

    def __init__(self, op: str, value: Decimal) -> None:
        if op == "=":
            self.match = lambda x: x == value
        elif op == ">=":
            self.match = lambda x: x >= value
        elif op == "<=":
            self.match = lambda x: x <= value
        elif op == ">":
            self.match = lambda x: x > value
        else:  # op == "<":
            self.match = lambda x: x < value

    def __call__(self, obj: Any) -> bool:
        # Compare to the absolute value to simplify this filter.
        number = getattr(obj, "number", None)
        return self.match(abs(number)) if number is not None else False

FilterSyntaxParser

Source code in src/rustfava/core/filters.py

class FilterSyntaxParser:
    precedence = (("left", "AND"), ("right", "UMINUS"))
    tokens = FilterSyntaxLexer.tokens

    def p_error(self, _: Any) -> None:
        raise FilterParseError

    def p_filter(self, p: list[Any]) -> None:
        """
        filter : expr
        """
        p[0] = p[1]

    def p_expr(self, p: list[Any]) -> None:
        """
        expr : simple_expr
        """
        p[0] = p[1]

    def p_expr_all(self, p: list[Any]) -> None:
        """
        expr : ALL expr ')'
        """
        expr = p[2]

        def _match_postings(entry: Directive) -> bool:
            return all(
                expr(posting) for posting in getattr(entry, "postings", [])
            )

        p[0] = _match_postings

    def p_expr_any(self, p: list[Any]) -> None:
        """
        expr : ANY expr ')'
        """
        expr = p[2]

        def _match_postings(entry: Directive) -> bool:
            return any(
                expr(posting) for posting in getattr(entry, "postings", [])
            )

        p[0] = _match_postings

    def p_expr_parentheses(self, p: list[Any]) -> None:
        """
        expr : '(' expr ')'
        """
        p[0] = p[2]

    def p_expr_and(self, p: list[Any]) -> None:
        """
        expr : expr expr %prec AND
        """
        left, right = p[1], p[2]

        def _and(entry: Directive) -> bool:
            return left(entry) and right(entry)  # type: ignore[no-any-return]

        p[0] = _and

    def p_expr_or(self, p: list[Any]) -> None:
        """
        expr : expr ',' expr
        """
        left, right = p[1], p[3]

        def _or(entry: Directive) -> bool:
            return left(entry) or right(entry)  # type: ignore[no-any-return]

        p[0] = _or

    def p_expr_negated(self, p: list[Any]) -> None:
        """
        expr : '-' expr %prec UMINUS
        """
        func = p[2]

        def _neg(entry: Directive) -> bool:
            return not func(entry)

        p[0] = _neg

    def p_simple_expr_TAG(self, p: list[Any]) -> None:  # noqa: N802
        """
        simple_expr : TAG
        """
        tag = p[1]

        def _tag(entry: Directive) -> bool:
            tags = getattr(entry, "tags", None)
            return (tag in tags) if tags is not None else False

        p[0] = _tag

    def p_simple_expr_LINK(self, p: list[Any]) -> None:  # noqa: N802
        """
        simple_expr : LINK
        """
        link = p[1]

        def _link(entry: Directive) -> bool:
            links = getattr(entry, "links", None)
            return (link in links) if links is not None else False

        p[0] = _link

    def p_simple_expr_STRING(self, p: list[Any]) -> None:  # noqa: N802
        """
        simple_expr : STRING
        """
        string = p[1]
        match = Match(string)

        def _string(entry: Directive) -> bool:
            for name in ("narration", "payee", "comment"):
                value = getattr(entry, name, "")
                if value and match(value):
                    return True
            return False

        p[0] = _string

    def p_simple_expr_key(self, p: list[Any]) -> None:
        """
        simple_expr : KEY EQ_OP STRING
                    | KEY CMP_OP NUMBER
        """
        key, op, value = p[1], p[2], p[3]
        match: Match | MatchAmount = (
            Match(value) if op == ":" else MatchAmount(op, value)
        )

        def _key(entry: Directive) -> bool:
            if hasattr(entry, key):
                return match(getattr(entry, key) or "")
            if entry.meta is not None and key in entry.meta:
                return match(entry.meta.get(key))
            return False

        p[0] = _key

    def p_simple_expr_units(self, p: list[Any]) -> None:
        """
        simple_expr : CMP_OP NUMBER
        """
        op, value = p[1], p[2]
        match = MatchAmount(op, value)

        def _range(entry: Directive) -> bool:
            return any(
                match(posting.units)
                for posting in getattr(entry, "postings", [])
            )

        p[0] = _range

p_filter(p)

filter : expr

Source code in src/rustfava/core/filters.py

def p_filter(self, p: list[Any]) -> None:
    """
    filter : expr
    """
    p[0] = p[1]

p_expr(p)

expr : simple_expr

Source code in src/rustfava/core/filters.py

def p_expr(self, p: list[Any]) -> None:
    """
    expr : simple_expr
    """
    p[0] = p[1]

p_expr_all(p)

expr : ALL expr ')'

Source code in src/rustfava/core/filters.py

def p_expr_all(self, p: list[Any]) -> None:
    """
    expr : ALL expr ')'
    """
    expr = p[2]

    def _match_postings(entry: Directive) -> bool:
        return all(
            expr(posting) for posting in getattr(entry, "postings", [])
        )

    p[0] = _match_postings

p_expr_any(p)

expr : ANY expr ')'

Source code in src/rustfava/core/filters.py

def p_expr_any(self, p: list[Any]) -> None:
    """
    expr : ANY expr ')'
    """
    expr = p[2]

    def _match_postings(entry: Directive) -> bool:
        return any(
            expr(posting) for posting in getattr(entry, "postings", [])
        )

    p[0] = _match_postings

p_expr_parentheses(p)

expr : '(' expr ')'

Source code in src/rustfava/core/filters.py

def p_expr_parentheses(self, p: list[Any]) -> None:
    """
    expr : '(' expr ')'
    """
    p[0] = p[2]

p_expr_and(p)

expr : expr expr %prec AND

Source code in src/rustfava/core/filters.py

def p_expr_and(self, p: list[Any]) -> None:
    """
    expr : expr expr %prec AND
    """
    left, right = p[1], p[2]

    def _and(entry: Directive) -> bool:
        return left(entry) and right(entry)  # type: ignore[no-any-return]

    p[0] = _and

p_expr_or(p)

expr : expr ',' expr

Source code in src/rustfava/core/filters.py

def p_expr_or(self, p: list[Any]) -> None:
    """
    expr : expr ',' expr
    """
    left, right = p[1], p[3]

    def _or(entry: Directive) -> bool:
        return left(entry) or right(entry)  # type: ignore[no-any-return]

    p[0] = _or

p_expr_negated(p)

expr : '-' expr %prec UMINUS

Source code in src/rustfava/core/filters.py

def p_expr_negated(self, p: list[Any]) -> None:
    """
    expr : '-' expr %prec UMINUS
    """
    func = p[2]

    def _neg(entry: Directive) -> bool:
        return not func(entry)

    p[0] = _neg

p_simple_expr_TAG(p)

simple_expr : TAG

Source code in src/rustfava/core/filters.py

def p_simple_expr_TAG(self, p: list[Any]) -> None:  # noqa: N802
    """
    simple_expr : TAG
    """
    tag = p[1]

    def _tag(entry: Directive) -> bool:
        tags = getattr(entry, "tags", None)
        return (tag in tags) if tags is not None else False

    p[0] = _tag

p_simple_expr_LINK(p)

simple_expr : LINK

Source code in src/rustfava/core/filters.py

def p_simple_expr_LINK(self, p: list[Any]) -> None:  # noqa: N802
    """
    simple_expr : LINK
    """
    link = p[1]

    def _link(entry: Directive) -> bool:
        links = getattr(entry, "links", None)
        return (link in links) if links is not None else False

    p[0] = _link

p_simple_expr_STRING(p)

simple_expr : STRING

Source code in src/rustfava/core/filters.py

def p_simple_expr_STRING(self, p: list[Any]) -> None:  # noqa: N802
    """
    simple_expr : STRING
    """
    string = p[1]
    match = Match(string)

    def _string(entry: Directive) -> bool:
        for name in ("narration", "payee", "comment"):
            value = getattr(entry, name, "")
            if value and match(value):
                return True
        return False

    p[0] = _string

p_simple_expr_key(p)

KEY EQ_OP STRING

| KEY CMP_OP NUMBER

Source code in src/rustfava/core/filters.py

def p_simple_expr_key(self, p: list[Any]) -> None:
    """
    simple_expr : KEY EQ_OP STRING
                | KEY CMP_OP NUMBER
    """
    key, op, value = p[1], p[2], p[3]
    match: Match | MatchAmount = (
        Match(value) if op == ":" else MatchAmount(op, value)
    )

    def _key(entry: Directive) -> bool:
        if hasattr(entry, key):
            return match(getattr(entry, key) or "")
        if entry.meta is not None and key in entry.meta:
            return match(entry.meta.get(key))
        return False

    p[0] = _key

p_simple_expr_units(p)

simple_expr : CMP_OP NUMBER

Source code in src/rustfava/core/filters.py

def p_simple_expr_units(self, p: list[Any]) -> None:
    """
    simple_expr : CMP_OP NUMBER
    """
    op, value = p[1], p[2]
    match = MatchAmount(op, value)

    def _range(entry: Directive) -> bool:
        return any(
            match(posting.units)
            for posting in getattr(entry, "postings", [])
        )

    p[0] = _range

EntryFilter

Bases: ABC

Filters a list of entries.

Source code in src/rustfava/core/filters.py

class EntryFilter(ABC):
    """Filters a list of entries."""

    @abstractmethod
    def apply(self, entries: Sequence[Directive]) -> Sequence[Directive]:
        """Filter a list of directives."""

apply(entries) abstractmethod

Filter a list of directives.

Source code in src/rustfava/core/filters.py

@abstractmethod
def apply(self, entries: Sequence[Directive]) -> Sequence[Directive]:
    """Filter a list of directives."""

TimeFilter

Bases: EntryFilter

Filter by dates.

Source code in src/rustfava/core/filters.py

class TimeFilter(EntryFilter):
    """Filter by dates."""

    __slots__ = ("date_range",)

    def __init__(
        self,
        options: BeancountOptions,
        fava_options: RustfavaOptions,
        value: str,
    ) -> None:
        del options  # unused
        begin, end = parse_date(value, fava_options.fiscal_year_end)
        if not begin or not end:
            raise TimeFilterParseError(value)
        self.date_range = DateRange(begin, end)

    def apply(self, entries: Sequence[Directive]) -> Sequence[Directive]:
        from rustfava.rustledger.engine import RustledgerEngine
        from rustfava.rustledger.types import directives_from_json
        from rustfava.rustledger.types import directives_to_json

        # Use native rustledger clamp_entries
        engine = RustledgerEngine.get_instance()
        entries_json = directives_to_json(list(entries))
        result = engine.clamp_entries(
            entries_json,
            str(self.date_range.begin),
            str(self.date_range.end),
        )
        return directives_from_json(result.get("entries", []))

AdvancedFilter

Bases: EntryFilter

Filter by tags and links and keys.

Source code in src/rustfava/core/filters.py

class AdvancedFilter(EntryFilter):
    """Filter by tags and links and keys."""

    __slots__ = ("_include",)

    def __init__(self, value: str) -> None:
        try:
            tokens = LEXER.lex(value)
            self._include = PARSE(
                lexer="NONE",
                tokenfunc=lambda toks=tokens: next(toks, None),  # ty:ignore[invalid-argument-type]
            )
        except FilterError as exception:
            exception.message += value
            raise

    def apply(self, entries: Sequence[Directive]) -> Sequence[Directive]:
        include = self._include
        return [entry for entry in entries if include(entry)]

AccountFilter

Bases: EntryFilter

Filter by account.

The filter string can either be a regular expression or a parent account.

Source code in src/rustfava/core/filters.py

class AccountFilter(EntryFilter):
    """Filter by account.

    The filter string can either be a regular expression or a parent account.
    """

    __slots__ = ("_match", "_value")

    def __init__(self, value: str) -> None:
        self._value = value
        self._match = Match(value)

    def apply(self, entries: Sequence[Directive]) -> Sequence[Directive]:
        value = self._value
        if not value:
            return entries
        match = self._match
        return [
            entry
            for entry in entries
            if any(
                _has_component(name, value) or match(name)
                for name in get_entry_accounts(entry)
            )
        ]

group_entries

Entries grouped by type.

EntriesByType

Bases: NamedTuple

Entries grouped by type.

Source code in src/rustfava/core/group_entries.py

class EntriesByType(NamedTuple):
    """Entries grouped by type."""

    Balance: Sequence[abc.Balance]
    Close: Sequence[abc.Close]
    Commodity: Sequence[abc.Commodity]
    Custom: Sequence[abc.Custom]
    Document: Sequence[abc.Document]
    Event: Sequence[abc.Event]
    Note: Sequence[abc.Note]
    Open: Sequence[abc.Open]
    Pad: Sequence[abc.Pad]
    Price: Sequence[abc.Price]
    Query: Sequence[abc.Query]
    Transaction: Sequence[abc.Transaction]

TransactionPosting

Bases: NamedTuple

Pair of a transaction and a posting.

Source code in src/rustfava/core/group_entries.py

class TransactionPosting(NamedTuple):
    """Pair of a transaction and a posting."""

    transaction: abc.Transaction
    posting: abc.Posting

group_entries_by_type(entries)

Group entries by type.

Parameters:

Name	Type	Description	Default
entries	Sequence[Directive]	A list of entries to group.	required

Returns:

Type	Description
EntriesByType	A namedtuple containing the grouped lists of entries.

Source code in src/rustfava/core/group_entries.py

def group_entries_by_type(entries: Sequence[abc.Directive]) -> EntriesByType:
    """Group entries by type.

    Arguments:
        entries: A list of entries to group.

    Returns:
        A namedtuple containing the grouped lists of entries.
    """
    entries_by_type = EntriesByType(
        [],
        [],
        [],
        [],
        [],
        [],
        [],
        [],
        [],
        [],
        [],
        [],
    )
    for entry in entries:
        # Handle both beancount types (e.g., "Transaction") and
        # rustledger types (e.g., "RLTransaction")
        type_name = entry.__class__.__name__
        if type_name.startswith("RL"):
            type_name = type_name[2:]  # Strip "RL" prefix
        getattr(entries_by_type, type_name).append(entry)
    return entries_by_type

group_entries_by_account(entries)

Group entries by account.

Parameters:

Name	Type	Description	Default
entries	Sequence[Directive]	A list of entries.	required

Returns:

Type	Description
Mapping[str, Sequence[Directive | TransactionPosting]]	A dict mapping account names to their entries.

Source code in src/rustfava/core/group_entries.py

def group_entries_by_account(
    entries: Sequence[abc.Directive],
) -> Mapping[str, Sequence[abc.Directive | TransactionPosting]]:
    """Group entries by account.

    Arguments:
        entries: A list of entries.

    Returns:
        A dict mapping account names to their entries.
    """
    res: dict[str, list[abc.Directive | TransactionPosting]] = defaultdict(
        list,
    )

    for entry in entries:
        if isinstance(entry, abc.Transaction):
            for posting in entry.postings:
                res[posting.account].append(TransactionPosting(entry, posting))
        else:
            for account in get_entry_accounts(entry):
                res[account].append(entry)

    return dict(sorted(res.items()))

ingest

Ingest helper functions.

IngestError dataclass

Bases: BeancountError

An error with one of the importers.

Source code in src/rustfava/core/ingest.py

class IngestError(BeancountError):
    """An error with one of the importers."""

ImporterMethodCallError

Bases: RustfavaAPIError

Error calling one of the importer methods.

Source code in src/rustfava/core/ingest.py

class ImporterMethodCallError(RustfavaAPIError):
    """Error calling one of the importer methods."""

    def __init__(self) -> None:
        super().__init__(
            f"Error calling method on importer:\n\n{traceback.format_exc()}"
        )

ImporterInvalidTypeError

Bases: RustfavaAPIError

One of the importer methods returned an unexpected type.

Source code in src/rustfava/core/ingest.py

class ImporterInvalidTypeError(RustfavaAPIError):
    """One of the importer methods returned an unexpected type."""

    def __init__(self, attr: str, expected: type[Any], actual: Any) -> None:
        super().__init__(
            f"Got unexpected type from importer as {attr}:"
            f" expected {expected!s}, got {type(actual)!s}:"
        )

ImporterExtractError

Bases: ImporterMethodCallError

Error calling extract for importer.

Source code in src/rustfava/core/ingest.py

class ImporterExtractError(ImporterMethodCallError):
    """Error calling extract for importer."""

MissingImporterConfigError

Bases: RustfavaAPIError

Missing import-config option.

Source code in src/rustfava/core/ingest.py

class MissingImporterConfigError(RustfavaAPIError):
    """Missing import-config option."""

    def __init__(self) -> None:
        super().__init__("Missing import-config option")

MissingImporterDirsError

Bases: RustfavaAPIError

You need to set at least one imports-dir.

Source code in src/rustfava/core/ingest.py

class MissingImporterDirsError(RustfavaAPIError):
    """You need to set at least one imports-dir."""

    def __init__(self) -> None:
        super().__init__("You need to set at least one imports-dir.")

ImportConfigLoadError

Bases: RustfavaAPIError

Error on loading the import config.

Source code in src/rustfava/core/ingest.py

class ImportConfigLoadError(RustfavaAPIError):
    """Error on loading the import config."""

FileImportInfo dataclass

Info about one file/importer combination.

Source code in src/rustfava/core/ingest.py

@dataclass(frozen=True)
class FileImportInfo:
    """Info about one file/importer combination."""

    importer_name: str
    account: str
    date: datetime.date
    name: str

FileImporters dataclass

Importers for a file.

Source code in src/rustfava/core/ingest.py

@dataclass(frozen=True)
class FileImporters:
    """Importers for a file."""

    name: str
    basename: str
    importers: list[FileImportInfo]

WrappedImporter

A wrapper to safely call importer methods.

Source code in src/rustfava/core/ingest.py

class WrappedImporter:
    """A wrapper to safely call importer methods."""

    importer: BeanImporterProtocol | Importer

    def __init__(self, importer: BeanImporterProtocol | Importer) -> None:
        self.importer = importer

    @property
    @_catch_any
    def name(self) -> str:
        """Get the name of the importer."""
        importer = self.importer
        name = (
            importer.name
            if isinstance(importer, Importer)
            else importer.name()
        )
        return _assert_type("name", name, str)

    @_catch_any
    def identify(self: WrappedImporter, path: Path) -> bool:
        """Whether the importer is matching the file."""
        importer = self.importer
        matches = (
            importer.identify(str(path))
            if isinstance(importer, Importer)
            else importer.identify(get_cached_file(path))
        )
        return _assert_type("identify", matches, bool)

    @_catch_any
    def file_import_info(self, path: Path) -> FileImportInfo:
        """Generate info about a file with an importer."""
        importer = self.importer
        if isinstance(importer, Importer):
            str_path = str(path)
            account = importer.account(str_path)
            date = importer.date(str_path)
            filename = importer.filename(str_path)
        else:
            file = get_cached_file(path)
            account = importer.file_account(file)
            date = importer.file_date(file)
            filename = importer.file_name(file)

        return FileImportInfo(
            self.name,
            _assert_type("account", account or "", str),
            _assert_type("date", date or local_today(), datetime.date),
            _assert_type("filename", filename or path.name, str),
        )

name property

Get the name of the importer.

identify(path)

Whether the importer is matching the file.

Source code in src/rustfava/core/ingest.py

@_catch_any
def identify(self: WrappedImporter, path: Path) -> bool:
    """Whether the importer is matching the file."""
    importer = self.importer
    matches = (
        importer.identify(str(path))
        if isinstance(importer, Importer)
        else importer.identify(get_cached_file(path))
    )
    return _assert_type("identify", matches, bool)

file_import_info(path)

Generate info about a file with an importer.

Source code in src/rustfava/core/ingest.py

@_catch_any
def file_import_info(self, path: Path) -> FileImportInfo:
    """Generate info about a file with an importer."""
    importer = self.importer
    if isinstance(importer, Importer):
        str_path = str(path)
        account = importer.account(str_path)
        date = importer.date(str_path)
        filename = importer.filename(str_path)
    else:
        file = get_cached_file(path)
        account = importer.file_account(file)
        date = importer.file_date(file)
        filename = importer.file_name(file)

    return FileImportInfo(
        self.name,
        _assert_type("account", account or "", str),
        _assert_type("date", date or local_today(), datetime.date),
        _assert_type("filename", filename or path.name, str),
    )

IngestModule

Bases: FavaModule

Exposes ingest functionality.

Source code in src/rustfava/core/ingest.py

class IngestModule(FavaModule):
    """Exposes ingest functionality."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        self.importers: Mapping[str, WrappedImporter] = {}
        self.hooks: Hooks = []
        self.mtime: int | None = None
        self.errors: list[IngestError] = []

    @property
    def module_path(self) -> Path | None:
        """The path to the importer configuration."""
        config_path = self.ledger.fava_options.import_config
        if not config_path:
            return None
        return self.ledger.join_path(config_path)

    def _error(self, msg: str) -> None:
        self.errors.append(
            IngestError(
                {"filename": str(self.module_path), "lineno": 0},
                msg,
                None,
            ),
        )

    def load_file(self) -> None:  # noqa: D102
        self.errors = []
        module_path = self.module_path
        if module_path is None:
            return

        if not module_path.exists():
            self._error("Import config does not exist")
            return

        new_mtime = module_path.stat().st_mtime_ns
        if new_mtime == self.mtime:
            return

        try:
            self.importers, self.hooks = load_import_config(module_path)
            self.mtime = new_mtime
        except RustfavaAPIError as error:  # pragma: no cover
            msg = f"Error in import config '{module_path}': {error!s}"
            self._error(msg)

    def import_data(self) -> list[FileImporters]:
        """Identify files and importers that can be imported.

        Returns:
            A list of :class:`.FileImportInfo`.
        """
        if not self.importers:
            return []

        importers = list(self.importers.values())

        ret: list[FileImporters] = []
        for directory in self.ledger.fava_options.import_dirs:
            full_path = self.ledger.join_path(directory)
            ret.extend(find_imports(importers, full_path))

        return ret

    def extract(self, filename: str, importer_name: str) -> list[Directive]:
        """Extract entries from filename with the specified importer.

        Args:
            filename: The full path to a file.
            importer_name: The name of an importer that matched the file.

        Returns:
            A list of new imported entries.
        """
        if not self.module_path:
            raise MissingImporterConfigError

        # reload (if changed)
        self.load_file()

        try:
            path = Path(filename)
            importer = self.importers[importer_name]
            new_entries = extract_from_file(
                importer,
                path,
                existing_entries=self.ledger.all_entries,
            )
        except Exception as exc:
            raise ImporterExtractError from exc

        for hook_fn in self.hooks:
            annotations = get_annotations(hook_fn)
            if any("Importer" in a for a in annotations.values()):
                importer_info = importer.file_import_info(path)
                new_entries_list: HookOutput = [
                    (
                        filename,
                        new_entries,
                        importer_info.account,
                        importer.importer,
                    )
                ]
            else:
                new_entries_list = [(filename, new_entries)]

            new_entries_list = hook_fn(
                new_entries_list,
                self.ledger.all_entries,
            )

            new_entries = new_entries_list[0][1]

        return new_entries

module_path property

The path to the importer configuration.

import_data()

Identify files and importers that can be imported.

Returns:

Type	Description
list[FileImporters]	A list of :class:.FileImportInfo.

Source code in src/rustfava/core/ingest.py

def import_data(self) -> list[FileImporters]:
    """Identify files and importers that can be imported.

    Returns:
        A list of :class:`.FileImportInfo`.
    """
    if not self.importers:
        return []

    importers = list(self.importers.values())

    ret: list[FileImporters] = []
    for directory in self.ledger.fava_options.import_dirs:
        full_path = self.ledger.join_path(directory)
        ret.extend(find_imports(importers, full_path))

    return ret

extract(filename, importer_name)

Extract entries from filename with the specified importer.

Parameters:

Name	Type	Description	Default
filename	str	The full path to a file.	required
importer_name	str	The name of an importer that matched the file.	required

Returns:

Type	Description
list[Directive]	A list of new imported entries.

Source code in src/rustfava/core/ingest.py

def extract(self, filename: str, importer_name: str) -> list[Directive]:
    """Extract entries from filename with the specified importer.

    Args:
        filename: The full path to a file.
        importer_name: The name of an importer that matched the file.

    Returns:
        A list of new imported entries.
    """
    if not self.module_path:
        raise MissingImporterConfigError

    # reload (if changed)
    self.load_file()

    try:
        path = Path(filename)
        importer = self.importers[importer_name]
        new_entries = extract_from_file(
            importer,
            path,
            existing_entries=self.ledger.all_entries,
        )
    except Exception as exc:
        raise ImporterExtractError from exc

    for hook_fn in self.hooks:
        annotations = get_annotations(hook_fn)
        if any("Importer" in a for a in annotations.values()):
            importer_info = importer.file_import_info(path)
            new_entries_list: HookOutput = [
                (
                    filename,
                    new_entries,
                    importer_info.account,
                    importer.importer,
                )
            ]
        else:
            new_entries_list = [(filename, new_entries)]

        new_entries_list = hook_fn(
            new_entries_list,
            self.ledger.all_entries,
        )

        new_entries = new_entries_list[0][1]

    return new_entries

walk_dir(directory)

Walk through all files in dir.

Ignores common dot-directories like .git, .cache. .venv, see IGNORE_DIRS.

Parameters:

Name	Type	Description	Default
directory	Path	The directory to start in.	required

Yields:

Type	Description
Iterable[Path]	All full paths under directory, ignoring some directories.

Source code in src/rustfava/core/ingest.py

def walk_dir(directory: Path) -> Iterable[Path]:
    """Walk through all files in dir.

    Ignores common dot-directories like .git, .cache. .venv, see IGNORE_DIRS.

    Args:
        directory: The directory to start in.

    Yields:
        All full paths under directory, ignoring some directories.
    """
    for root, dirs, filenames in os.walk(directory):
        dirs[:] = sorted(d for d in dirs if d not in IGNORE_DIRS)
        root_path = Path(root)
        for filename in sorted(filenames):
            yield root_path / filename

get_cached_file(path)

Get a cached FileMemo.

This checks the file's mtime before getting it from the Cache. In addition to using the beangulp cache.

Source code in src/rustfava/core/ingest.py

def get_cached_file(path: Path) -> FileMemo:
    """Get a cached FileMemo.

    This checks the file's mtime before getting it from the Cache.
    In addition to using the beangulp cache.
    """
    mtime = path.stat().st_mtime_ns
    filename = str(path)
    cached = _CACHE.get(path)
    if cached:
        mtime_cached, memo_cached = cached
        if mtime <= mtime_cached:  # pragma: no cover
            return memo_cached
    memo: FileMemo = cache._FileMemo(filename)  # noqa: SLF001
    cache._CACHE[filename] = memo  # noqa: SLF001
    _CACHE[path] = (mtime, memo)
    return memo

find_imports(config, directory)

Pair files and matching importers.

Yields:

Type	Description
Iterable[FileImporters]	For each file in directory, a pair of its filename and the matching
Iterable[FileImporters]	importers.

Source code in src/rustfava/core/ingest.py

def find_imports(
    config: Sequence[WrappedImporter], directory: Path
) -> Iterable[FileImporters]:
    """Pair files and matching importers.

    Yields:
        For each file in directory, a pair of its filename and the matching
        importers.
    """
    for path in walk_dir(directory):
        stat = path.stat()
        if stat.st_size > _FILE_TOO_LARGE_THRESHOLD:  # pragma: no cover
            continue

        importers = [
            importer.file_import_info(path)
            for importer in config
            if importer.identify(path)
        ]
        yield FileImporters(
            name=str(path), basename=path.name, importers=importers
        )

extract_from_file(wrapped_importer, path, existing_entries)

Import entries from a document.

Parameters:

Name	Type	Description	Default
wrapped_importer	WrappedImporter	The importer instance to handle the document.	required
path	Path	Filesystem path to the document.	required
existing_entries	Sequence[Directive]	Existing entries.	required

Returns:

Type	Description
list[Directive]	The list of imported entries.

Source code in src/rustfava/core/ingest.py

def extract_from_file(
    wrapped_importer: WrappedImporter,
    path: Path,
    existing_entries: Sequence[Directive],
) -> list[Directive]:
    """Import entries from a document.

    Args:
      wrapped_importer: The importer instance to handle the document.
      path: Filesystem path to the document.
      existing_entries: Existing entries.

    Returns:
      The list of imported entries.
    """
    filename = str(path)
    importer = wrapped_importer.importer
    if isinstance(importer, Importer):
        entries = importer.extract(filename, existing=existing_entries)
    else:
        file = get_cached_file(path)
        entries = (
            importer.extract(file, existing_entries=existing_entries)
            if "existing_entries" in signature(importer.extract).parameters
            else importer.extract(file)
        ) or []

    if hasattr(importer, "sort"):
        importer.sort(entries)
    else:
        entries.sort(key=_incomplete_sortkey)
    if isinstance(importer, Importer):
        importer.deduplicate(entries, existing=existing_entries)
    return entries

load_import_config(module_path)

Load the given import config and extract importers and hooks.

Parameters:

Name	Type	Description	Default
module_path	Path	Path to the import config.	required

Returns:

Type	Description
tuple[Mapping[str, WrappedImporter], Hooks]	A pair of the importers (by name) and the list of hooks.

Source code in src/rustfava/core/ingest.py

def load_import_config(
    module_path: Path,
) -> tuple[Mapping[str, WrappedImporter], Hooks]:
    """Load the given import config and extract importers and hooks.

    Args:
        module_path: Path to the import config.

    Returns:
        A pair of the importers (by name) and the list of hooks.
    """
    try:
        mod = run_path(str(module_path))
    except Exception as error:  # pragma: no cover
        message = traceback.format_exc()
        raise ImportConfigLoadError(message) from error

    if "CONFIG" not in mod:
        msg = "CONFIG is missing"
        raise ImportConfigLoadError(msg)
    if not isinstance(mod["CONFIG"], list):  # pragma: no cover
        msg = "CONFIG is not a list"
        raise ImportConfigLoadError(msg)

    config = mod["CONFIG"]
    hooks = DEFAULT_HOOKS
    if "HOOKS" in mod:  # pragma: no cover
        hooks = mod["HOOKS"]
        if not isinstance(hooks, list) or not all(
            callable(fn) for fn in hooks
        ):
            msg = "HOOKS is not a list of callables"
            raise ImportConfigLoadError(msg)
    importers = {}
    for importer in config:
        if not isinstance(
            importer, (BeanImporterProtocol, Importer)
        ):  # pragma: no cover
            name = importer.__class__.__name__
            msg = (
                f"Importer class '{name}' in '{module_path}' does "
                "not satisfy importer protocol"
            )
            raise ImportConfigLoadError(msg)
        wrapped_importer = WrappedImporter(importer)
        if wrapped_importer.name in importers:
            msg = f"Duplicate importer name found: {wrapped_importer.name}"
            raise ImportConfigLoadError(msg)
        importers[wrapped_importer.name] = wrapped_importer
    return importers, hooks

filepath_in_primary_imports_folder(filename, ledger)

File path for a document to upload to the primary import folder.

Parameters:

Name	Type	Description	Default
filename	str	The filename of the document.	required
ledger	RustfavaLedger	The RustfavaLedger.	required

Returns:

Type	Description
Path	The path that the document should be saved at.

Source code in src/rustfava/core/ingest.py

def filepath_in_primary_imports_folder(
    filename: str,
    ledger: RustfavaLedger,
) -> Path:
    """File path for a document to upload to the primary import folder.

    Args:
        filename: The filename of the document.
        ledger: The RustfavaLedger.

    Returns:
        The path that the document should be saved at.
    """
    primary_imports_folder = next(iter(ledger.fava_options.import_dirs), None)
    if primary_imports_folder is None:
        raise MissingImporterDirsError

    filename = filename.replace(sep, " ")
    if altsep:  # pragma: no cover
        filename = filename.replace(altsep, " ")

    return ledger.join_path(primary_imports_folder, filename)

inventory

Alternative implementation of Beancount's Inventory.

SimpleCounterInventory

Bases: dict[str, Decimal]

A simple inventory mapping just strings to numbers.

Source code in src/rustfava/core/inventory.py

class SimpleCounterInventory(dict[str, Decimal]):
    """A simple inventory mapping just strings to numbers."""

    def is_empty(self) -> bool:
        """Check if the inventory is empty."""
        return not bool(self)

    def add(self, key: str, number: Decimal) -> None:
        """Add a number to key."""
        new_num = number + self.get(key, ZERO)
        if new_num == ZERO:
            self.pop(key, None)
        else:
            self[key] = new_num

    def __iter__(self) -> Iterator[str]:
        raise NotImplementedError

    def __neg__(self) -> SimpleCounterInventory:
        return SimpleCounterInventory({key: -num for key, num in self.items()})

    def reduce(
        self,
        reducer: Callable[Concatenate[Position, P], Amount],
        *args: P.args,
        **_kwargs: P.kwargs,
    ) -> SimpleCounterInventory:
        """Reduce inventory."""
        counter = SimpleCounterInventory()
        for currency, number in self.items():
            pos = _Position(_Amount(number, currency), None)
            amount = reducer(pos, *args)  # type: ignore[call-arg]
            counter.add(amount.currency, amount.number)
        return counter

is_empty()

Check if the inventory is empty.

Source code in src/rustfava/core/inventory.py

def is_empty(self) -> bool:
    """Check if the inventory is empty."""
    return not bool(self)

add(key, number)

Add a number to key.

Source code in src/rustfava/core/inventory.py

def add(self, key: str, number: Decimal) -> None:
    """Add a number to key."""
    new_num = number + self.get(key, ZERO)
    if new_num == ZERO:
        self.pop(key, None)
    else:
        self[key] = new_num

reduce(reducer, *args, **_kwargs)

Reduce inventory.

Source code in src/rustfava/core/inventory.py

def reduce(
    self,
    reducer: Callable[Concatenate[Position, P], Amount],
    *args: P.args,
    **_kwargs: P.kwargs,
) -> SimpleCounterInventory:
    """Reduce inventory."""
    counter = SimpleCounterInventory()
    for currency, number in self.items():
        pos = _Position(_Amount(number, currency), None)
        amount = reducer(pos, *args)  # type: ignore[call-arg]
        counter.add(amount.currency, amount.number)
    return counter

CounterInventory

Bases: dict[InventoryKey, Decimal]

A lightweight inventory.

This is intended as a faster alternative to Beancount's Inventory class. Due to not using a list, for inventories with a lot of different positions, inserting is much faster.

The keys should be tuples (currency, cost).

Source code in src/rustfava/core/inventory.py

class CounterInventory(dict[InventoryKey, Decimal]):
    """A lightweight inventory.

    This is intended as a faster alternative to Beancount's Inventory class.
    Due to not using a list, for inventories with a lot of different positions,
    inserting is much faster.

    The keys should be tuples ``(currency, cost)``.
    """

    def is_empty(self) -> bool:
        """Check if the inventory is empty."""
        return not bool(self)

    def add(self, key: InventoryKey, number: Decimal) -> None:
        """Add a number to key."""
        new_num = number + self.get(key, ZERO)
        if new_num == ZERO:
            self.pop(key, None)
        else:
            self[key] = new_num

    def __iter__(self) -> Iterator[InventoryKey]:
        raise NotImplementedError

    def to_strings(self) -> list[str]:
        """Print as a list of strings (e.g. for snapshot tests)."""
        strings = []
        for (currency, cost), number in self.items():
            if cost is None:
                strings.append(f"{number} {currency}")
            else:
                cost_str = cost_to_string(cost)
                strings.append(f"{number} {currency} {{{cost_str}}}")
        return strings

    def reduce(
        self,
        reducer: Callable[Concatenate[Position, P], Amount],
        *args: P.args,
        **_kwargs: P.kwargs,
    ) -> SimpleCounterInventory:
        """Reduce inventory.

        Note that this returns a simple :class:`CounterInventory` with just
        currencies as keys.
        """
        counter = SimpleCounterInventory()
        for (currency, cost), number in self.items():
            pos = _Position(_Amount(number, currency), cost)
            amount = reducer(pos, *args)  # type: ignore[call-arg]
            counter.add(amount.currency, amount.number)
        return counter

    def add_amount(self, amount: Amount, cost: Cost | None = None) -> None:
        """Add an Amount to the inventory."""
        key = (amount.currency, cost)
        self.add(key, amount.number)

    def add_position(self, pos: Position) -> None:
        """Add a Position or Posting to the inventory."""
        # Skip positions with missing units (can happen with parse errors)
        if pos.units is None:
            return
        self.add_amount(pos.units, pos.cost)

    def __neg__(self) -> CounterInventory:
        return CounterInventory({key: -num for key, num in self.items()})

    def __add__(self, other: CounterInventory) -> CounterInventory:
        counter = CounterInventory(self)
        counter.add_inventory(other)
        return counter

    def add_inventory(self, counter: CounterInventory) -> None:
        """Add another :class:`CounterInventory`."""
        if not self:
            self.update(counter)
        else:
            self_get = self.get
            for key, num in counter.items():
                new_num = num + self_get(key, ZERO)
                if new_num == ZERO:
                    self.pop(key, None)
                else:
                    self[key] = new_num

is_empty()

Check if the inventory is empty.

Source code in src/rustfava/core/inventory.py

def is_empty(self) -> bool:
    """Check if the inventory is empty."""
    return not bool(self)

add(key, number)

Add a number to key.

Source code in src/rustfava/core/inventory.py

def add(self, key: InventoryKey, number: Decimal) -> None:
    """Add a number to key."""
    new_num = number + self.get(key, ZERO)
    if new_num == ZERO:
        self.pop(key, None)
    else:
        self[key] = new_num

to_strings()

Print as a list of strings (e.g. for snapshot tests).

Source code in src/rustfava/core/inventory.py

def to_strings(self) -> list[str]:
    """Print as a list of strings (e.g. for snapshot tests)."""
    strings = []
    for (currency, cost), number in self.items():
        if cost is None:
            strings.append(f"{number} {currency}")
        else:
            cost_str = cost_to_string(cost)
            strings.append(f"{number} {currency} {{{cost_str}}}")
    return strings

reduce(reducer, *args, **_kwargs)

Reduce inventory.

Note that this returns a simple :class:CounterInventory with just currencies as keys.

Source code in src/rustfava/core/inventory.py

def reduce(
    self,
    reducer: Callable[Concatenate[Position, P], Amount],
    *args: P.args,
    **_kwargs: P.kwargs,
) -> SimpleCounterInventory:
    """Reduce inventory.

    Note that this returns a simple :class:`CounterInventory` with just
    currencies as keys.
    """
    counter = SimpleCounterInventory()
    for (currency, cost), number in self.items():
        pos = _Position(_Amount(number, currency), cost)
        amount = reducer(pos, *args)  # type: ignore[call-arg]
        counter.add(amount.currency, amount.number)
    return counter

add_amount(amount, cost=None)

Add an Amount to the inventory.

Source code in src/rustfava/core/inventory.py

def add_amount(self, amount: Amount, cost: Cost | None = None) -> None:
    """Add an Amount to the inventory."""
    key = (amount.currency, cost)
    self.add(key, amount.number)

add_position(pos)

Add a Position or Posting to the inventory.

Source code in src/rustfava/core/inventory.py

def add_position(self, pos: Position) -> None:
    """Add a Position or Posting to the inventory."""
    # Skip positions with missing units (can happen with parse errors)
    if pos.units is None:
        return
    self.add_amount(pos.units, pos.cost)

add_inventory(counter)

Add another :class:CounterInventory.

Source code in src/rustfava/core/inventory.py

def add_inventory(self, counter: CounterInventory) -> None:
    """Add another :class:`CounterInventory`."""
    if not self:
        self.update(counter)
    else:
        self_get = self.get
        for key, num in counter.items():
            new_num = num + self_get(key, ZERO)
            if new_num == ZERO:
                self.pop(key, None)
            else:
                self[key] = new_num

misc

Some miscellaneous reports.

FavaError dataclass

Bases: BeancountError

Generic Fava-specific error.

Source code in src/rustfava/core/misc.py

class FavaError(BeancountError):
    """Generic Fava-specific error."""

FavaMisc

Bases: FavaModule

Provides access to some miscellaneous reports.

Source code in src/rustfava/core/misc.py

class FavaMisc(FavaModule):
    """Provides access to some miscellaneous reports."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        #: User-chosen links to show in the sidebar.
        self.sidebar_links: SidebarLinks = []
        #: Upcoming events in the next few days.
        self.upcoming_events: Sequence[Event] = []

    def load_file(self) -> None:  # noqa: D102
        custom_entries = self.ledger.all_entries_by_type.Custom
        self.sidebar_links = sidebar_links(custom_entries)

        self.upcoming_events = upcoming_events(
            self.ledger.all_entries_by_type.Event,
            self.ledger.fava_options.upcoming_events,
        )

    @property
    def errors(self) -> Sequence[FavaError]:
        """An error if no operating currency is set."""
        return (
            []
            if self.ledger.options["operating_currency"]
            else [NO_OPERATING_CURRENCY_ERROR]
        )

errors property

An error if no operating currency is set.

sidebar_links(custom_entries)

Parse custom entries for links.

They have the following format:

2016-04-01 custom "fava-sidebar-link" "2014" "/income_statement/?time=2014"

Source code in src/rustfava/core/misc.py

def sidebar_links(custom_entries: Sequence[Custom]) -> SidebarLinks:
    """Parse custom entries for links.

    They have the following format:

    2016-04-01 custom "fava-sidebar-link" "2014" "/income_statement/?time=2014"
    """
    sidebar_link_entries = [
        entry for entry in custom_entries if entry.type == "fava-sidebar-link"
    ]
    return [
        (entry.values[0].value, entry.values[1].value)
        for entry in sidebar_link_entries
    ]

upcoming_events(events, max_delta)

Parse entries for upcoming events.

Parameters:

Name	Type	Description	Default
events	Sequence[Event]	A list of events.	required
max_delta	int	Number of days that should be considered.	required

Returns:

Type	Description
Sequence[Event]	A list of the Events in entries that are less than max_delta days
Sequence[Event]	away.

Source code in src/rustfava/core/misc.py

def upcoming_events(
    events: Sequence[Event], max_delta: int
) -> Sequence[Event]:
    """Parse entries for upcoming events.

    Args:
        events: A list of events.
        max_delta: Number of days that should be considered.

    Returns:
        A list of the Events in entries that are less than `max_delta` days
        away.
    """
    today = local_today()
    upcoming = []

    for event in events:
        delta = event.date - today
        if delta.days >= 0 and delta.days < max_delta:
            upcoming.append(event)

    return upcoming

align(string, currency_column)

Align currencies in one column.

Source code in src/rustfava/beans/str.py

def align(string: str, currency_column: int) -> str:
    """Align currencies in one column."""
    output = io.StringIO()
    for line in string.splitlines():
        match = ALIGN_RE.match(line)
        if match:
            prefix, number, rest = match.groups()
            num_of_spaces = currency_column - len(prefix) - len(number) - 4
            spaces = " " * num_of_spaces
            output.write(prefix + spaces + "  " + number + " " + rest)
        else:
            output.write(line)
        output.write("\n")

    return output.getvalue()

module_base

Base class for the "modules" of rustfavaLedger.

FavaModule

Base class for the "modules" of rustfavaLedger.

Source code in src/rustfava/core/module_base.py

class FavaModule:
    """Base class for the "modules" of rustfavaLedger."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        self.ledger = ledger

    def load_file(self) -> None:
        """Run when the file has been (re)loaded."""

load_file()

Run when the file has been (re)loaded.

Source code in src/rustfava/core/module_base.py

def load_file(self) -> None:
    """Run when the file has been (re)loaded."""

number

Formatting numbers.

DecimalFormatModule

Bases: FavaModule

Formatting numbers.

Source code in src/rustfava/core/number.py

class DecimalFormatModule(FavaModule):
    """Formatting numbers."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        self._locale: Locale | None = None
        self._formatters: dict[str, Formatter] = {}
        self._default_pattern = get_locale_format(None, 2)
        self.precisions: dict[str, int] = {}

    def load_file(self) -> None:  # noqa: D102
        locale = None

        locale_option = self.ledger.fava_options.locale
        if (
            self.ledger.options["render_commas"] and not locale_option
        ):  # pragma: no cover
            locale_option = "en"
            self.ledger.fava_options.locale = locale_option

        if locale_option:
            locale = Locale.parse(locale_option)

        dcontext = self.ledger.options["dcontext"]
        precisions: dict[str, int] = {}

        # Both beancount's DisplayContext and RLDisplayContext have ccontexts
        for currency, ccontext in dcontext.ccontexts.items():
            prec = ccontext.get_fractional(None)
            if prec is not None:
                precisions[currency] = prec

        precisions.update(self.ledger.commodities.precisions)

        self._locale = locale
        self._default_pattern = get_locale_format(locale, 2)
        self._formatters = {
            currency: get_locale_format(locale, prec)
            for currency, prec in precisions.items()
        }
        self.precisions = precisions

    def __call__(self, value: Decimal, currency: str | None = None) -> str:
        """Format a decimal to the right number of decimal digits with locale.

        Arguments:
            value: A decimal number.
            currency: A currency string or None.

        Returns:
            A string, the formatted decimal.
        """
        if currency is None:
            return self._default_pattern(value)
        return self._formatters.get(currency, self._default_pattern)(value)

__call__(value, currency=None)

Format a decimal to the right number of decimal digits with locale.

Parameters:

Name	Type	Description	Default
value	Decimal	A decimal number.	required
currency	str | None	A currency string or None.	None

Returns:

Type	Description
str	A string, the formatted decimal.

Source code in src/rustfava/core/number.py

def __call__(self, value: Decimal, currency: str | None = None) -> str:
    """Format a decimal to the right number of decimal digits with locale.

    Arguments:
        value: A decimal number.
        currency: A currency string or None.

    Returns:
        A string, the formatted decimal.
    """
    if currency is None:
        return self._default_pattern(value)
    return self._formatters.get(currency, self._default_pattern)(value)

get_locale_format(locale, precision)

Obtain formatting pattern for the given locale and precision.

Parameters:

Name	Type	Description	Default
locale	Locale | None	An optional locale.	required
precision	int	The precision.	required

Returns:

Type	Description
Formatter	A function that renders Decimals to strings as desired.

Source code in src/rustfava/core/number.py

def get_locale_format(locale: Locale | None, precision: int) -> Formatter:
    """Obtain formatting pattern for the given locale and precision.

    Arguments:
        locale: An optional locale.
        precision: The precision.

    Returns:
        A function that renders Decimals to strings as desired.
    """
    # Set a maximum precision of 14, half the default precision of Decimal
    precision = min(precision, 14)
    if locale is None:
        fmt_string = "{:." + str(precision) + "f}"

        def fmt(num: Decimal) -> str:
            return fmt_string.format(num)

        return fmt

    pattern = copy.copy(locale.decimal_formats.get(None))
    if not pattern:  # pragma: no cover
        msg = "Expected Locale to have a decimal format pattern"
        raise ValueError(msg)
    pattern.frac_prec = (precision, precision)

    def locale_fmt(num: Decimal) -> str:
        return pattern.apply(num, locale)  # type: ignore[no-any-return]

    return locale_fmt

query

Query result types.

QueryResultTable dataclass

Table query result.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class QueryResultTable:
    """Table query result."""

    types: list[BaseColumn]
    rows: list[tuple[SerialisedQueryRowValue, ...]]
    t: Literal["table"] = "table"

QueryResultText dataclass

Text query result.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class QueryResultText:
    """Text query result."""

    contents: str
    t: Literal["string"] = "string"

BaseColumn dataclass

A query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class BaseColumn:
    """A query column."""

    name: str
    dtype: str

    @staticmethod
    def serialise(
        val: QueryRowValue,
    ) -> SerialisedQueryRowValue:
        """Serialiseable version of the column value."""
        return val  # type: ignore[no-any-return]

serialise(val) staticmethod

Serialiseable version of the column value.

Source code in src/rustfava/core/query.py

@staticmethod
def serialise(
    val: QueryRowValue,
) -> SerialisedQueryRowValue:
    """Serialiseable version of the column value."""
    return val  # type: ignore[no-any-return]

BoolColumn dataclass

Bases: BaseColumn

A boolean query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class BoolColumn(BaseColumn):
    """A boolean query column."""

    dtype: str = "bool"

DecimalColumn dataclass

Bases: BaseColumn

A Decimal query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class DecimalColumn(BaseColumn):
    """A Decimal query column."""

    dtype: str = "Decimal"

IntColumn dataclass

Bases: BaseColumn

A int query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class IntColumn(BaseColumn):
    """A int query column."""

    dtype: str = "int"

StrColumn dataclass

Bases: BaseColumn

A str query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class StrColumn(BaseColumn):
    """A str query column."""

    dtype: str = "str"

DateColumn dataclass

Bases: BaseColumn

A date query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class DateColumn(BaseColumn):
    """A date query column."""

    dtype: str = "date"

PositionColumn dataclass

Bases: BaseColumn

A Position query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class PositionColumn(BaseColumn):
    """A Position query column."""

    dtype: str = "Position"

SetColumn dataclass

Bases: BaseColumn

A set query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class SetColumn(BaseColumn):
    """A set query column."""

    dtype: str = "set"

AmountColumn dataclass

Bases: BaseColumn

An amount query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class AmountColumn(BaseColumn):
    """An amount query column."""

    dtype: str = "Amount"

ObjectColumn dataclass

Bases: BaseColumn

An object query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class ObjectColumn(BaseColumn):
    """An object query column."""

    dtype: str = "object"

    @staticmethod
    def serialise(val: object) -> str:
        """Serialise an object of unknown type to a string."""
        return str(val)

serialise(val) staticmethod

Serialise an object of unknown type to a string.

Source code in src/rustfava/core/query.py

@staticmethod
def serialise(val: object) -> str:
    """Serialise an object of unknown type to a string."""
    return str(val)

InventoryColumn dataclass

Bases: BaseColumn

An inventory query column.

Source code in src/rustfava/core/query.py

@dataclass(frozen=True)
class InventoryColumn(BaseColumn):
    """An inventory query column."""

    dtype: str = "Inventory"

    @staticmethod
    def serialise(
        val: dict[str, Decimal] | None,
    ) -> SimpleCounterInventory | None:
        """Serialise an inventory.

        Rustledger returns inventory as a dict of currency -> Decimal.
        """
        if val is None:
            return None
        # Rustledger already converts to {currency: Decimal} format
        if isinstance(val, dict):
            from rustfava.core.inventory import SimpleCounterInventory
            return SimpleCounterInventory(val)
        # Fallback for beancount Inventory type (for backwards compat)
        return UNITS.apply_inventory(val) if val is not None else None

serialise(val) staticmethod

Serialise an inventory.

Rustledger returns inventory as a dict of currency -> Decimal.

Source code in src/rustfava/core/query.py

@staticmethod
def serialise(
    val: dict[str, Decimal] | None,
) -> SimpleCounterInventory | None:
    """Serialise an inventory.

    Rustledger returns inventory as a dict of currency -> Decimal.
    """
    if val is None:
        return None
    # Rustledger already converts to {currency: Decimal} format
    if isinstance(val, dict):
        from rustfava.core.inventory import SimpleCounterInventory
        return SimpleCounterInventory(val)
    # Fallback for beancount Inventory type (for backwards compat)
    return UNITS.apply_inventory(val) if val is not None else None

query_shell

For running BQL queries in Fava.

FavaShellError

Bases: RustfavaAPIError

An error in the Fava BQL shell, will be turned into a string.

Source code in src/rustfava/core/query_shell.py

class FavaShellError(RustfavaAPIError):
    """An error in the Fava BQL shell, will be turned into a string."""

QueryNotFoundError

Bases: FavaShellError

Query '{name}' not found.

Source code in src/rustfava/core/query_shell.py

class QueryNotFoundError(FavaShellError):
    """Query '{name}' not found."""

    def __init__(self, name: str) -> None:
        super().__init__(f"Query '{name}' not found.")

TooManyRunArgsError

Bases: FavaShellError

Too many args to run: '{args}'.

Source code in src/rustfava/core/query_shell.py

class TooManyRunArgsError(FavaShellError):
    """Too many args to run: '{args}'."""

    def __init__(self, args: str) -> None:
        super().__init__(f"Too many args to run: '{args}'.")

QueryCompilationError

Bases: FavaShellError

Query compilation error.

Source code in src/rustfava/core/query_shell.py

class QueryCompilationError(FavaShellError):
    """Query compilation error."""

    def __init__(self, err: CompilationError) -> None:
        super().__init__(f"Query compilation error: {err!s}.")

QueryParseError

Bases: FavaShellError

Query parse error.

Source code in src/rustfava/core/query_shell.py

class QueryParseError(FavaShellError):
    """Query parse error."""

    def __init__(self, err: ParseError) -> None:
        super().__init__(f"Query parse error: {err!s}.")

NonExportableQueryError

Bases: FavaShellError

Only queries that return a table can be printed to a file.

Source code in src/rustfava/core/query_shell.py

class NonExportableQueryError(FavaShellError):
    """Only queries that return a table can be printed to a file."""

    def __init__(self) -> None:
        super().__init__(
            "Only queries that return a table can be printed to a file."
        )

FavaQueryRunner

Runs BQL queries using rustledger.

Source code in src/rustfava/core/query_shell.py

class FavaQueryRunner:
    """Runs BQL queries using rustledger."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        self.ledger = ledger

    def run(
        self, entries: Sequence[Directive], query: str
    ) -> RLCursor | str:
        """Run a query, returning cursor or text result."""
        # Get the source from the ledger for queries
        source = getattr(self.ledger, "_source", None)

        # Create connection
        conn = connect(
            "rustledger:",
            entries=entries,
            errors=self.ledger.errors,
            options=self.ledger.options,
        )

        if source:
            conn.set_source(source)

        # Parse the query to handle special commands
        query = query.strip()
        query_lower = query.lower()

        # Handle noop commands (return fixed text)
        noop_doc = "Doesn't do anything in rustfava's query shell."
        if query_lower in (".exit", ".quit", "exit", "quit"):
            return noop_doc

        # Handle .run or run command
        if query_lower.startswith((".run", "run")):
            # Check if it's just "run" or ".run" (list queries) or "run name"
            if query_lower in ("run", ".run") or query_lower.startswith(("run ", ".run ")):
                return self._handle_run(query, conn)

        # Handle help commands - return text
        if query_lower.startswith((".help", "help")):
            # ".help exit" or ".help <command>" returns noop doc
            if " " in query_lower:
                return noop_doc
            return self._help_text()

        # Handle .explain - return placeholder
        if query_lower.startswith((".explain", "explain")):
            return f"EXPLAIN: {query}"

        # Handle SELECT/BALANCES/JOURNAL queries
        try:
            return conn.execute(query)
        except ParseError as exc:
            raise QueryParseError(exc) from exc
        except CompilationError as exc:
            raise QueryCompilationError(exc) from exc

    def _handle_run(self, query: str, conn: RLConnection) -> RLCursor | str:
        """Handle .run command to execute stored queries."""
        queries = self.ledger.all_entries_by_type.Query

        # Parse the run command
        parts = shlex.split(query)
        if len(parts) == 1:
            # Just "run" - list available queries
            return "\n".join(q.name for q in queries)

        if len(parts) > 2:
            raise TooManyRunArgsError(query)

        name = parts[1].rstrip(";")
        query_obj = next((q for q in queries if q.name == name), None)
        if query_obj is None:
            raise QueryNotFoundError(name)

        try:
            return conn.execute(query_obj.query_string)
        except ParseError as exc:
            raise QueryParseError(exc) from exc
        except CompilationError as exc:
            raise QueryCompilationError(exc) from exc

    def _help_text(self) -> str:
        """Return help text for the query shell."""
        return """Fava Query Shell

Commands:
  SELECT ...     Run a BQL SELECT query
  run <name>     Run a stored query by name
  run            List all stored queries
  help           Show this help message

Example queries:
  SELECT account, sum(position) GROUP BY account
  SELECT date, narration, position WHERE account ~ "Expenses"
"""

run(entries, query)

Run a query, returning cursor or text result.

Source code in src/rustfava/core/query_shell.py

def run(
    self, entries: Sequence[Directive], query: str
) -> RLCursor | str:
    """Run a query, returning cursor or text result."""
    # Get the source from the ledger for queries
    source = getattr(self.ledger, "_source", None)

    # Create connection
    conn = connect(
        "rustledger:",
        entries=entries,
        errors=self.ledger.errors,
        options=self.ledger.options,
    )

    if source:
        conn.set_source(source)

    # Parse the query to handle special commands
    query = query.strip()
    query_lower = query.lower()

    # Handle noop commands (return fixed text)
    noop_doc = "Doesn't do anything in rustfava's query shell."
    if query_lower in (".exit", ".quit", "exit", "quit"):
        return noop_doc

    # Handle .run or run command
    if query_lower.startswith((".run", "run")):
        # Check if it's just "run" or ".run" (list queries) or "run name"
        if query_lower in ("run", ".run") or query_lower.startswith(("run ", ".run ")):
            return self._handle_run(query, conn)

    # Handle help commands - return text
    if query_lower.startswith((".help", "help")):
        # ".help exit" or ".help <command>" returns noop doc
        if " " in query_lower:
            return noop_doc
        return self._help_text()

    # Handle .explain - return placeholder
    if query_lower.startswith((".explain", "explain")):
        return f"EXPLAIN: {query}"

    # Handle SELECT/BALANCES/JOURNAL queries
    try:
        return conn.execute(query)
    except ParseError as exc:
        raise QueryParseError(exc) from exc
    except CompilationError as exc:
        raise QueryCompilationError(exc) from exc

QueryShell

Bases: FavaModule

A Fava module to run BQL queries.

Source code in src/rustfava/core/query_shell.py

class QueryShell(FavaModule):
    """A Fava module to run BQL queries."""

    def __init__(self, ledger: RustfavaLedger) -> None:
        super().__init__(ledger)
        self.runner = FavaQueryRunner(ledger)

    def execute_query_serialised(
        self, entries: Sequence[Directive], query: str
    ) -> QueryResultTable | QueryResultText:
        """Run a query and returns its serialised result.

        Arguments:
            entries: The entries to run the query on.
            query: A query string.

        Returns:
            Either a table or a text result (depending on the query).

        Raises:
            RustfavaAPIError: If the query response is an error.
        """
        res = self.runner.run(entries, query)
        return (
            QueryResultText(res) if isinstance(res, str) else _serialise(res)
        )

    def query_to_file(
        self,
        entries: Sequence[Directive],
        query_string: str,
        result_format: str,
    ) -> tuple[str, io.BytesIO]:
        """Get query result as file.

        Arguments:
            entries: The entries to run the query on.
            query_string: A string, the query to run.
            result_format: The file format to save to.

        Returns:
            A tuple (name, data), where name is either 'query_result' or the
            name of a custom query if the query string is 'run name_of_query'.
            ``data`` contains the file contents.

        Raises:
            RustfavaAPIError: If the result format is not supported or the
            query failed.
        """
        name = "query_result"

        if query_string.lower().startswith((".run", "run ")):
            parts = shlex.split(query_string)
            if len(parts) > 2:
                raise TooManyRunArgsError(query_string)
            if len(parts) == 2:
                name = parts[1].rstrip(";")
                queries = self.ledger.all_entries_by_type.Query
                query_obj = next((q for q in queries if q.name == name), None)
                if query_obj is None:
                    raise QueryNotFoundError(name)
                query_string = query_obj.query_string

        res = self.runner.run(entries, query_string)
        if isinstance(res, str):
            raise NonExportableQueryError

        rrows = res.fetchall()
        rtypes = res.description

        # Convert rows to exportable format
        rows = _numberify_rows(rrows, rtypes)

        if result_format == "csv":
            data = to_csv(list(rtypes), rows)
        else:
            if not HAVE_EXCEL:  # pragma: no cover
                msg = "Result format not supported."
                raise RustfavaAPIError(msg)
            data = to_excel(list(rtypes), rows, result_format, query_string)
        return name, data

execute_query_serialised(entries, query)

Run a query and returns its serialised result.

Parameters:

Name	Type	Description	Default
entries	Sequence[Directive]	The entries to run the query on.	required
query	str	A query string.	required

Returns:

Type	Description
QueryResultTable | QueryResultText	Either a table or a text result (depending on the query).

Raises:

Type	Description
RustfavaAPIError	If the query response is an error.

Source code in src/rustfava/core/query_shell.py

def execute_query_serialised(
    self, entries: Sequence[Directive], query: str
) -> QueryResultTable | QueryResultText:
    """Run a query and returns its serialised result.

    Arguments:
        entries: The entries to run the query on.
        query: A query string.

    Returns:
        Either a table or a text result (depending on the query).

    Raises:
        RustfavaAPIError: If the query response is an error.
    """
    res = self.runner.run(entries, query)
    return (
        QueryResultText(res) if isinstance(res, str) else _serialise(res)
    )

query_to_file(entries, query_string, result_format)

Get query result as file.

Parameters:

Name	Type	Description	Default
entries	Sequence[Directive]	The entries to run the query on.	required
query_string	str	A string, the query to run.	required
result_format	str	The file format to save to.	required

Returns:

Type	Description
str	A tuple (name, data), where name is either 'query_result' or the
BytesIO	name of a custom query if the query string is 'run name_of_query'.
tuple[str, BytesIO]	data contains the file contents.

Raises:

Type	Description
RustfavaAPIError	If the result format is not supported or the

Source code in src/rustfava/core/query_shell.py

def query_to_file(
    self,
    entries: Sequence[Directive],
    query_string: str,
    result_format: str,
) -> tuple[str, io.BytesIO]:
    """Get query result as file.

    Arguments:
        entries: The entries to run the query on.
        query_string: A string, the query to run.
        result_format: The file format to save to.

    Returns:
        A tuple (name, data), where name is either 'query_result' or the
        name of a custom query if the query string is 'run name_of_query'.
        ``data`` contains the file contents.

    Raises:
        RustfavaAPIError: If the result format is not supported or the
        query failed.
    """
    name = "query_result"

    if query_string.lower().startswith((".run", "run ")):
        parts = shlex.split(query_string)
        if len(parts) > 2:
            raise TooManyRunArgsError(query_string)
        if len(parts) == 2:
            name = parts[1].rstrip(";")
            queries = self.ledger.all_entries_by_type.Query
            query_obj = next((q for q in queries if q.name == name), None)
            if query_obj is None:
                raise QueryNotFoundError(name)
            query_string = query_obj.query_string

    res = self.runner.run(entries, query_string)
    if isinstance(res, str):
        raise NonExportableQueryError

    rrows = res.fetchall()
    rtypes = res.description

    # Convert rows to exportable format
    rows = _numberify_rows(rrows, rtypes)

    if result_format == "csv":
        data = to_csv(list(rtypes), rows)
    else:
        if not HAVE_EXCEL:  # pragma: no cover
            msg = "Result format not supported."
            raise RustfavaAPIError(msg)
        data = to_excel(list(rtypes), rows, result_format, query_string)
    return name, data

tree

Account balance trees.

SerialisedTreeNode dataclass

A serialised TreeNode.

Source code in src/rustfava/core/tree.py

@dataclass(frozen=True)
class SerialisedTreeNode:
    """A serialised TreeNode."""

    account: str
    balance: SimpleCounterInventory
    balance_children: SimpleCounterInventory
    children: Sequence[SerialisedTreeNode]
    has_txns: bool
    cost: SimpleCounterInventory | None = None
    cost_children: SimpleCounterInventory | None = None

TreeNode

A node in the account tree.

Source code in src/rustfava/core/tree.py

class TreeNode:
    """A node in the account tree."""

    __slots__ = ("balance", "balance_children", "children", "has_txns", "name")

    def __init__(self, name: str) -> None:
        #: Account name.
        self.name: str = name
        #: A list of :class:`.TreeNode`, its children.
        self.children: list[TreeNode] = []
        #: The cumulative account balance.
        self.balance_children = CounterInventory()
        #: The account balance.
        self.balance = CounterInventory()
        #: Whether the account has any transactions.
        self.has_txns = False

    def serialise(
        self,
        conversion: Conversion,
        prices: RustfavaPriceMap,
        end: datetime.date | None,
        *,
        with_cost: bool = False,
    ) -> SerialisedTreeNode:
        """Serialise the account.

        Args:
            conversion: The conversion to use.
            prices: The price map to use.
            end: A date to use for cost conversions.
            with_cost: Additionally convert to cost.
        """
        children = [
            child.serialise(conversion, prices, end, with_cost=with_cost)
            for child in sorted(self.children, key=attrgetter("name"))
        ]
        return (
            SerialisedTreeNode(
                self.name,
                conversion.apply(self.balance, prices, end),
                conversion.apply(self.balance_children, prices, end),
                children,
                self.has_txns,
                AT_COST.apply(self.balance),
                AT_COST.apply(self.balance_children),
            )
            if with_cost
            else SerialisedTreeNode(
                self.name,
                conversion.apply(self.balance, prices, end),
                conversion.apply(self.balance_children, prices, end),
                children,
                self.has_txns,
            )
        )

    def serialise_with_context(self) -> SerialisedTreeNode:
        """Serialise, getting all parameters from Flask context."""
        return self.serialise(
            g.conv,
            g.ledger.prices,
            g.filtered.end_date,
            with_cost=g.conv == AT_VALUE,
        )

serialise(conversion, prices, end, *, with_cost=False)

Serialise the account.

Parameters:

Name	Type	Description	Default
conversion	Conversion	The conversion to use.	required
prices	RustfavaPriceMap	The price map to use.	required
end	date | None	A date to use for cost conversions.	required
with_cost	bool	Additionally convert to cost.	False

Source code in src/rustfava/core/tree.py

def serialise(
    self,
    conversion: Conversion,
    prices: RustfavaPriceMap,
    end: datetime.date | None,
    *,
    with_cost: bool = False,
) -> SerialisedTreeNode:
    """Serialise the account.

    Args:
        conversion: The conversion to use.
        prices: The price map to use.
        end: A date to use for cost conversions.
        with_cost: Additionally convert to cost.
    """
    children = [
        child.serialise(conversion, prices, end, with_cost=with_cost)
        for child in sorted(self.children, key=attrgetter("name"))
    ]
    return (
        SerialisedTreeNode(
            self.name,
            conversion.apply(self.balance, prices, end),
            conversion.apply(self.balance_children, prices, end),
            children,
            self.has_txns,
            AT_COST.apply(self.balance),
            AT_COST.apply(self.balance_children),
        )
        if with_cost
        else SerialisedTreeNode(
            self.name,
            conversion.apply(self.balance, prices, end),
            conversion.apply(self.balance_children, prices, end),
            children,
            self.has_txns,
        )
    )

serialise_with_context()

Serialise, getting all parameters from Flask context.

Source code in src/rustfava/core/tree.py

def serialise_with_context(self) -> SerialisedTreeNode:
    """Serialise, getting all parameters from Flask context."""
    return self.serialise(
        g.conv,
        g.ledger.prices,
        g.filtered.end_date,
        with_cost=g.conv == AT_VALUE,
    )

Tree

Bases: dict[str, TreeNode]

Account tree.

Parameters:

Name	Type	Description	Default
entries	Iterable[Directive | Directive] | None	A list of entries to compute balances from.	None
create_accounts	list[str] | None	A list of accounts that the tree should contain.	None

Source code in src/rustfava/core/tree.py

class Tree(dict[str, TreeNode]):
    """Account tree.

    Args:
        entries: A list of entries to compute balances from.
        create_accounts: A list of accounts that the tree should contain.
    """

    def __init__(
        self,
        entries: Iterable[Directive | data.Directive] | None = None,
        create_accounts: list[str] | None = None,
    ) -> None:
        super().__init__(self)
        self.get("", insert=True)
        if create_accounts:
            for account in create_accounts:
                self.get(account, insert=True)
        if entries:
            account_balances: dict[str, CounterInventory]
            account_balances = defaultdict(CounterInventory)
            for entry in entries:
                if isinstance(entry, Open):
                    self.get(entry.account, insert=True)
                for posting in getattr(entry, "postings", []):
                    account_balances[posting.account].add_position(posting)

            for name, balance in sorted(account_balances.items()):
                self.insert(name, balance)

    @property
    def accounts(self) -> list[str]:
        """The accounts in this tree."""
        return sorted(self.keys())

    def ancestors(self, name: str) -> Iterable[TreeNode]:
        """Ancestors of an account.

        Args:
            name: An account name.

        Yields:
            The ancestors of the given account from the bottom up.
        """
        while name:
            name = account_parent(name) or ""
            yield self.get(name)

    def insert(self, name: str, balance: CounterInventory) -> None:
        """Insert account with a balance.

        Insert account and update its balance and the balances of its
        ancestors.

        Args:
            name: An account name.
            balance: The balance of the account.
        """
        node = self.get(name, insert=True)
        node.balance.add_inventory(balance)
        node.balance_children.add_inventory(balance)
        node.has_txns = True
        for parent_node in self.ancestors(name):
            parent_node.balance_children.add_inventory(balance)

    def get(  # type: ignore[override]
        self,
        name: str,
        *,
        insert: bool = False,
    ) -> TreeNode:
        """Get an account.

        Args:
            name: An account name.
            insert: If True, insert the name into the tree if it does not
                exist.

        Returns:
            TreeNode: The account of that name or an empty account if the
            account is not in the tree.
        """
        try:
            return self[name]
        except KeyError:
            node = TreeNode(name)
            if insert:
                if name:
                    parent = self.get(account_parent(name) or "", insert=True)
                    parent.children.append(node)
                self[name] = node
            return node

    def net_profit(
        self,
        options: BeancountOptions,
        account_name: str,
    ) -> TreeNode:
        """Calculate the net profit.

        Args:
            options: The Beancount options.
            account_name: The name to use for the account containing the net
                profit.
        """
        income = self.get(options["name_income"])
        expenses = self.get(options["name_expenses"])

        net_profit = Tree()
        net_profit.insert(
            account_name,
            income.balance_children + expenses.balance_children,
        )

        return net_profit.get(account_name)

    def cap(self, options: BeancountOptions, unrealized_account: str) -> None:
        """Transfer Income and Expenses, add conversions and unrealized gains.

        Args:
            options: The Beancount options.
            unrealized_account: The name of the account to post unrealized
                gains to (as a subaccount of Equity).
        """
        equity = options["name_equity"]
        conversions = CounterInventory(
            {
                (currency, None): -number
                for currency, number in AT_COST.apply(
                    self.get("").balance_children
                ).items()
            },
        )

        # Add conversions
        self.insert(
            equity + ":" + options["account_current_conversions"],
            conversions,
        )

        # Insert unrealized gains.
        self.insert(
            equity + ":" + unrealized_account,
            -self.get("").balance_children,
        )

        # Transfer Income and Expenses
        self.insert(
            equity + ":" + options["account_current_earnings"],
            self.get(options["name_income"]).balance_children,
        )
        self.insert(
            equity + ":" + options["account_current_earnings"],
            self.get(options["name_expenses"]).balance_children,
        )

accounts property

The accounts in this tree.

ancestors(name)

Ancestors of an account.

Parameters:

Name	Type	Description	Default
name	str	An account name.	required

Yields:

Type	Description
Iterable[TreeNode]	The ancestors of the given account from the bottom up.

Source code in src/rustfava/core/tree.py

def ancestors(self, name: str) -> Iterable[TreeNode]:
    """Ancestors of an account.

    Args:
        name: An account name.

    Yields:
        The ancestors of the given account from the bottom up.
    """
    while name:
        name = account_parent(name) or ""
        yield self.get(name)

insert(name, balance)

Insert account with a balance.

Insert account and update its balance and the balances of its ancestors.

Parameters:

Name	Type	Description	Default
name	str	An account name.	required
balance	CounterInventory	The balance of the account.	required

Source code in src/rustfava/core/tree.py

def insert(self, name: str, balance: CounterInventory) -> None:
    """Insert account with a balance.

    Insert account and update its balance and the balances of its
    ancestors.

    Args:
        name: An account name.
        balance: The balance of the account.
    """
    node = self.get(name, insert=True)
    node.balance.add_inventory(balance)
    node.balance_children.add_inventory(balance)
    node.has_txns = True
    for parent_node in self.ancestors(name):
        parent_node.balance_children.add_inventory(balance)

get(name, *, insert=False)

Get an account.

Parameters:

Name	Type	Description	Default
name	str	An account name.	required
insert	bool	If True, insert the name into the tree if it does not exist.	False

Returns:

Name	Type	Description
TreeNode	TreeNode	The account of that name or an empty account if the
	TreeNode	account is not in the tree.

Source code in src/rustfava/core/tree.py

def get(  # type: ignore[override]
    self,
    name: str,
    *,
    insert: bool = False,
) -> TreeNode:
    """Get an account.

    Args:
        name: An account name.
        insert: If True, insert the name into the tree if it does not
            exist.

    Returns:
        TreeNode: The account of that name or an empty account if the
        account is not in the tree.
    """
    try:
        return self[name]
    except KeyError:
        node = TreeNode(name)
        if insert:
            if name:
                parent = self.get(account_parent(name) or "", insert=True)
                parent.children.append(node)
            self[name] = node
        return node

net_profit(options, account_name)

Calculate the net profit.

Parameters:

Name	Type	Description	Default
options	BeancountOptions	The Beancount options.	required
account_name	str	The name to use for the account containing the net profit.	required

Source code in src/rustfava/core/tree.py

def net_profit(
    self,
    options: BeancountOptions,
    account_name: str,
) -> TreeNode:
    """Calculate the net profit.

    Args:
        options: The Beancount options.
        account_name: The name to use for the account containing the net
            profit.
    """
    income = self.get(options["name_income"])
    expenses = self.get(options["name_expenses"])

    net_profit = Tree()
    net_profit.insert(
        account_name,
        income.balance_children + expenses.balance_children,
    )

    return net_profit.get(account_name)

cap(options, unrealized_account)

Transfer Income and Expenses, add conversions and unrealized gains.

Parameters:

Name	Type	Description	Default
options	BeancountOptions	The Beancount options.	required
unrealized_account	str	The name of the account to post unrealized gains to (as a subaccount of Equity).	required

Source code in src/rustfava/core/tree.py

def cap(self, options: BeancountOptions, unrealized_account: str) -> None:
    """Transfer Income and Expenses, add conversions and unrealized gains.

    Args:
        options: The Beancount options.
        unrealized_account: The name of the account to post unrealized
            gains to (as a subaccount of Equity).
    """
    equity = options["name_equity"]
    conversions = CounterInventory(
        {
            (currency, None): -number
            for currency, number in AT_COST.apply(
                self.get("").balance_children
            ).items()
        },
    )

    # Add conversions
    self.insert(
        equity + ":" + options["account_current_conversions"],
        conversions,
    )

    # Insert unrealized gains.
    self.insert(
        equity + ":" + unrealized_account,
        -self.get("").balance_children,
    )

    # Transfer Income and Expenses
    self.insert(
        equity + ":" + options["account_current_earnings"],
        self.get(options["name_income"]).balance_children,
    )
    self.insert(
        equity + ":" + options["account_current_earnings"],
        self.get(options["name_expenses"]).balance_children,
    )

watcher

A simple file and folder watcher.

WatcherBase

Bases: ABC

ABC for rustfava ledger file watchers.

Source code in src/rustfava/core/watcher.py

class WatcherBase(abc.ABC):
    """ABC for rustfava ledger file watchers."""

    last_checked: int
    """Timestamp of the latest change noticed by the file watcher."""

    last_notified: int
    """Timestamp of the latest change that the watcher was notified of."""

    @abc.abstractmethod
    def update(self, files: Iterable[Path], folders: Iterable[Path]) -> None:
        """Update the folders/files to watch.

        Args:
            files: A list of file paths.
            folders: A list of paths to folders.
        """

    def check(self) -> bool:
        """Check for changes.

        Returns:
            `True` if there was a file change in one of the files or folders,
            `False` otherwise.
        """
        latest_mtime = max(self._get_latest_mtime(), self.last_notified)
        has_higher_mtime = latest_mtime > self.last_checked
        if has_higher_mtime:
            self.last_checked = latest_mtime
        return has_higher_mtime

    def notify(self, path: Path) -> None:
        """Notify the watcher of a change to a path."""
        try:
            change_mtime = Path(path).stat().st_mtime_ns
        except FileNotFoundError:
            change_mtime = max(self.last_notified, self.last_checked) + 1
        self.last_notified = max(self.last_notified, change_mtime)

    @abc.abstractmethod
    def _get_latest_mtime(self) -> int:
        """Get the latest change mtime."""

last_checked instance-attribute

Timestamp of the latest change noticed by the file watcher.

last_notified instance-attribute

Timestamp of the latest change that the watcher was notified of.

update(files, folders) abstractmethod

Update the folders/files to watch.

Parameters:

Name	Type	Description	Default
files	Iterable[Path]	A list of file paths.	required
folders	Iterable[Path]	A list of paths to folders.	required

Source code in src/rustfava/core/watcher.py

@abc.abstractmethod
def update(self, files: Iterable[Path], folders: Iterable[Path]) -> None:
    """Update the folders/files to watch.

    Args:
        files: A list of file paths.
        folders: A list of paths to folders.
    """

check()

Check for changes.

Returns:

Type	Description
bool	True if there was a file change in one of the files or folders,
bool	False otherwise.

Source code in src/rustfava/core/watcher.py

def check(self) -> bool:
    """Check for changes.

    Returns:
        `True` if there was a file change in one of the files or folders,
        `False` otherwise.
    """
    latest_mtime = max(self._get_latest_mtime(), self.last_notified)
    has_higher_mtime = latest_mtime > self.last_checked
    if has_higher_mtime:
        self.last_checked = latest_mtime
    return has_higher_mtime

notify(path)

Notify the watcher of a change to a path.

Source code in src/rustfava/core/watcher.py

def notify(self, path: Path) -> None:
    """Notify the watcher of a change to a path."""
    try:
        change_mtime = Path(path).stat().st_mtime_ns
    except FileNotFoundError:
        change_mtime = max(self.last_notified, self.last_checked) + 1
    self.last_notified = max(self.last_notified, change_mtime)

WatchfilesWatcher

Bases: WatcherBase

A file and folder watcher using the watchfiles library.

Source code in src/rustfava/core/watcher.py

class WatchfilesWatcher(WatcherBase):
    """A file and folder watcher using the watchfiles library."""

    def __init__(self) -> None:
        self.last_checked = 0
        self.last_notified = 0
        self._paths: tuple[set[Path], set[Path]] | None = None
        self._watchers: tuple[_WatchfilesThread, _WatchfilesThread] | None = (
            None
        )

    def update(self, files: Iterable[Path], folders: Iterable[Path]) -> None:
        """Update the folders/files to watch."""
        files_set = {p.absolute() for p in files if p.exists()}
        folders_set = {p.absolute() for p in folders if p.is_dir()}
        new_paths = (files_set, folders_set)
        if self._watchers and new_paths == self._paths:
            self.check()
            return
        self._paths = new_paths
        if self._watchers:
            self._watchers[0].stop()
            self._watchers[1].stop()
        self._watchers = (
            _FilesWatchfilesThread(files_set, self.last_checked),
            _WatchfilesThread(folders_set, self.last_checked, recursive=True),
        )
        self._watchers[0].start()
        self._watchers[1].start()
        self.check()

    def __enter__(self) -> None:
        pass

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: types.TracebackType | None,
    ) -> None:
        if self._watchers:
            self._watchers[0].stop()
            self._watchers[1].stop()

    def _get_latest_mtime(self) -> int:
        return (
            max(self._watchers[0].mtime, self._watchers[1].mtime)
            if self._watchers
            else 0
        )

update(files, folders)

Update the folders/files to watch.

Source code in src/rustfava/core/watcher.py

def update(self, files: Iterable[Path], folders: Iterable[Path]) -> None:
    """Update the folders/files to watch."""
    files_set = {p.absolute() for p in files if p.exists()}
    folders_set = {p.absolute() for p in folders if p.is_dir()}
    new_paths = (files_set, folders_set)
    if self._watchers and new_paths == self._paths:
        self.check()
        return
    self._paths = new_paths
    if self._watchers:
        self._watchers[0].stop()
        self._watchers[1].stop()
    self._watchers = (
        _FilesWatchfilesThread(files_set, self.last_checked),
        _WatchfilesThread(folders_set, self.last_checked, recursive=True),
    )
    self._watchers[0].start()
    self._watchers[1].start()
    self.check()

Watcher

Bases: WatcherBase

A simple file and folder watcher.

For folders, only checks mtime of the folder and all subdirectories. So a file change won't be noticed, but only new/deleted files.

Source code in src/rustfava/core/watcher.py

class Watcher(WatcherBase):
    """A simple file and folder watcher.

    For folders, only checks mtime of the folder and all subdirectories.
    So a file change won't be noticed, but only new/deleted files.
    """

    def __init__(self) -> None:
        self.last_checked = 0
        self.last_notified = 0
        self._files: Sequence[Path] = []
        self._folders: Sequence[Path] = []

    def update(self, files: Iterable[Path], folders: Iterable[Path]) -> None:
        """Update the folders/files to watch."""
        self._files = list(files)
        self._folders = list(folders)
        self.check()

    def _mtimes(self) -> Iterable[int]:
        for path in self._files:
            try:
                yield path.stat().st_mtime_ns
            except FileNotFoundError:
                yield max(self.last_notified, self.last_checked) + 1
        for path in self._folders:
            for dirpath, _, _ in walk(path):
                yield Path(dirpath).stat().st_mtime_ns

    def _get_latest_mtime(self) -> int:
        return max(self._mtimes())

update(files, folders)

Update the folders/files to watch.

Source code in src/rustfava/core/watcher.py

def update(self, files: Iterable[Path], folders: Iterable[Path]) -> None:
    """Update the folders/files to watch."""
    self._files = list(files)
    self._folders = list(folders)
    self.check()

Application

rustfava.application

rustfava's main WSGI application.

you can use create_app to create a rustfava WSGI app for a given list of files. To start a simple server::

from rustfava.application import create_app

app = create_app(['/path/to/file.beancount'])
app.run('localhost', 5000)

static_url(filename)

Return a static url with an mtime query string for cache busting.

Source code in src/rustfava/application.py

def static_url(filename: str) -> str:
    """Return a static url with an mtime query string for cache busting."""
    file_path = Path(__file__).parent / "static" / filename
    try:
        mtime = str(int(file_path.stat().st_mtime))
    except FileNotFoundError:
        mtime = "0"
    return url_for("static", filename=filename, mtime=mtime)

url_for(endpoint, **values)

Wrap flask.url_for using a cache.

Source code in src/rustfava/application.py

def url_for(endpoint: str, **values: str) -> str:
    """Wrap flask.url_for using a cache."""
    _inject_filters(endpoint, values)
    return _cached_url_for(endpoint, **values)

translations()

Get translations catalog.

Source code in src/rustfava/application.py

def translations() -> dict[str, str]:
    """Get translations catalog."""
    catalog = get_translations()._catalog  # noqa: SLF001
    return {k: v for k, v in catalog.items() if isinstance(k, str) and k}

create_app(files, *, load=False, incognito=False, read_only=False, poll_watcher=False)

Create a rustfava Flask application.

Parameters:

Name	Type	Description	Default
files	Iterable[Path | str]	The list of Beancount files (paths).	required
load	bool	Whether to load the Beancount files directly.	False
incognito	bool	Whether to run in incognito mode.	False
read_only	bool	Whether to run in read-only mode.	False
poll_watcher	bool	Whether to use old poll watcher	False

Source code in src/rustfava/application.py

def create_app(
    files: Iterable[Path | str],
    *,
    load: bool = False,
    incognito: bool = False,
    read_only: bool = False,
    poll_watcher: bool = False,
) -> Flask:
    """Create a rustfava Flask application.

    Arguments:
        files: The list of Beancount files (paths).
        load: Whether to load the Beancount files directly.
        incognito: Whether to run in incognito mode.
        read_only: Whether to run in read-only mode.
        poll_watcher: Whether to use old poll watcher
    """
    fava_app = Flask("rustfava")
    fava_app.register_blueprint(json_api, url_prefix="/<bfile>/api")
    fava_app.json = RustfavaJSONProvider(fava_app)
    fava_app.app_ctx_globals_class = Context  # type: ignore[assignment]
    _setup_template_config(fava_app, incognito=incognito)
    _setup_babel(fava_app)
    _setup_filters(fava_app, read_only=read_only)
    _setup_routes(fava_app)

    fava_app.config["HAVE_EXCEL"] = HAVE_EXCEL
    fava_app.config["BEANCOUNT_FILES"] = [str(f) for f in files]
    fava_app.config["INCOGNITO"] = incognito
    fava_app.config["LEDGERS"] = _LedgerSlugLoader(
        fava_app, load=load, poll_watcher=poll_watcher
    )

    return fava_app

CLI

rustfava.cli

The command-line interface for rustfava.

main(*, filenames=(), port=5000, host='localhost', prefix=None, incognito=False, read_only=False, debug=False, profile=False, profile_dir=None, poll_watcher=False)

Start Rustfava for FILENAMES on http://:.

If the BEANCOUNT_FILE environment variable is set, Rustfava will use the files (delimited by ';' on Windows and ':' on POSIX) given there in addition to FILENAMES.

Note you can also specify command-line options via environment variables with the RUSTFAVA_ prefix. For example, --host=0.0.0.0 is equivalent to setting the environment variable RUSTFAVA_HOST=0.0.0.0.

Source code in src/rustfava/cli.py

@click.command(context_settings={"auto_envvar_prefix": "RUSTFAVA"})
@click.argument(
    "filenames",
    nargs=-1,
    type=click.Path(exists=True, dir_okay=False, resolve_path=True),
)
@click.option(
    "-p",
    "--port",
    type=int,
    default=5000,
    show_default=True,
    metavar="<port>",
    help="The port to listen on.",
)
@click.option(
    "-H",
    "--host",
    type=str,
    default="localhost",
    show_default=True,
    metavar="<host>",
    help="The host to listen on.",
)
@click.option("--prefix", type=str, help="Set an URL prefix.")
@click.option(
    "--incognito",
    is_flag=True,
    help="Run in incognito mode and obscure all numbers.",
)
@click.option(
    "--read-only",
    is_flag=True,
    help="Run in read-only mode, disable any change through rustfava.",
)
@click.option("-d", "--debug", is_flag=True, help="Turn on debugging.")
@click.option(
    "--profile",
    is_flag=True,
    help="Turn on profiling. Implies --debug.",
)
@click.option(
    "--profile-dir",
    type=click.Path(),
    help="Output directory for profiling data.",
)
@click.option(
    "--poll-watcher", is_flag=True, help="Use old polling-based watcher."
)
@click.version_option(package_name="rustfava")
def main(  # noqa: PLR0913
    *,
    filenames: tuple[str, ...] = (),
    port: int = 5000,
    host: str = "localhost",
    prefix: str | None = None,
    incognito: bool = False,
    read_only: bool = False,
    debug: bool = False,
    profile: bool = False,
    profile_dir: str | None = None,
    poll_watcher: bool = False,
) -> None:  # pragma: no cover
    """Start Rustfava for FILENAMES on http://<host>:<port>.

    If the `BEANCOUNT_FILE` environment variable is set, Rustfava will use the
    files (delimited by ';' on Windows and ':' on POSIX) given there in
    addition to FILENAMES.

    Note you can also specify command-line options via environment variables
    with the `RUSTFAVA_` prefix. For example, `--host=0.0.0.0` is equivalent to
    setting the environment variable `RUSTFAVA_HOST=0.0.0.0`.
    """
    all_filenames = _add_env_filenames(filenames)

    if not all_filenames:
        raise NoFileSpecifiedError

    from rustfava.application import create_app

    app = create_app(
        all_filenames,
        incognito=incognito,
        read_only=read_only,
        poll_watcher=poll_watcher,
    )

    if prefix:
        from werkzeug.middleware.dispatcher import DispatcherMiddleware

        from rustfava.util import simple_wsgi

        app.wsgi_app = DispatcherMiddleware(  # type: ignore[method-assign]
            simple_wsgi,
            {prefix: app.wsgi_app},
        )

    # ensure that cheroot does not use IP6 for localhost
    host = "127.0.0.1" if host == "localhost" else host
    # Debug mode if profiling is active
    debug = debug or profile

    click.secho(f"Starting Fava on http://{host}:{port}", fg="green")
    if not debug:
        from cheroot.wsgi import Server

        server = Server((host, port), app)
        try:
            server.start()
        except KeyboardInterrupt:
            click.echo("Keyboard interrupt received: stopping Fava", err=True)
            server.stop()
        except OSError as error:
            if "No socket could be created" in str(error):
                raise AddressInUse(port) from error
            raise click.Abort from error
    else:
        from werkzeug.middleware.profiler import ProfilerMiddleware

        from rustfava.util import setup_debug_logging

        setup_debug_logging()
        if profile:
            app.wsgi_app = ProfilerMiddleware(  # type: ignore[method-assign]
                app.wsgi_app,
                restrictions=(30,),
                profile_dir=profile_dir or None,
            )

        app.jinja_env.auto_reload = True
        try:
            app.run(host, port, debug)
        except OSError as error:
            if error.errno == errno.EADDRINUSE:
                raise AddressInUse(port) from error
            raise