File policy_lib_test.py has 514 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import decimal
import textwrap
from typing import Any, Callable, Iterator, Optional, TypeVar
from beancount.parser import parser
from beancount.core.data import Balance, Custom, Directive, Transaction

File split_account.py has 445 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import collections
import dataclasses
import decimal
import functools
import itertools

File importer.py has 409 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import datetime
from decimal import Decimal
import http.server
import logging
import os

Function deduplicate has a Cognitive Complexity of 28 (exceeds 5 allowed). Consider refactoring.

Open

def deduplicate(
        new_entries: list[Directive],
        existing_entries: list[Directive],
        window_days: int = 10) -> list[Directive]:
    """De-duplicate entries.

• Read up
  Read up

Function check_file_entries has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

def check_file_entries(entries: list[Directive]) -> list[error_lib.Error]:
    """Checks entries are in order and finds out-of-order entries.

    We find a longest non-descending subsequence and assumes all other
    entries are out-of-order.

• Read up
  Read up

Function plugin has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

def plugin(entries: list[Directive], options: dict[str, Any]) -> tuple[list[Directive], list[error_lib.Error]]:
    entries_by_file: defaultdict[str | None, list[Directive]] = defaultdict(list)
    ignored_files: set[str] = set()
    errors: list[error_lib.Error] = []
    for entry in entries:

• Read up
  Read up

Function try_parse_policy_definition has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

def try_parse_policy_definition(meta: Optional[dict[str, Any]]) -> Optional[PolicyDefinition]:
    if not meta:
        return None
    weights = {}
    for key, value in meta.items():

• Read up
  Read up

Function _get_args has a Cognitive Complexity of 17 (exceeds 5 allowed). Consider refactoring.

Open

def _get_args(
        values: list[grammar.ValueType],
        handler: _CustomHandler,
) -> Optional[list[Any]]:
    queue = values[::-1]

• Read up
  Read up

Function _get_arg has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

def _get_arg(annotation: Any, queue: list[grammar.ValueType]) -> Optional[Any]:
    if not queue:
        return None
    value = queue.pop()
    if get_origin(annotation) is Union:

• Read up
  Read up

Function _process_entry has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def _process_entry(self, entry: Directive) -> Iterator[Directive]:
        if isinstance(entry, Custom) and (handlers := self._CUSTOM_HANDLERS.get(entry.type)):
            for chandler in handlers:
                if (args := _get_args(entry.values or [], chandler)) is not None:
                    with _wrap_plugin_exception(entry, self._error_logger):

• Read up
  Read up

Function load_test_suite has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

def load_test_suite(tests_path: str, suite: str) -> tuple[list[str], list[Testcase]]:
    full_path = os.path.join(tests_path, suite)
    ids = list[str]()
    testcases = list[Testcase]()
    if not os.path.isdir(full_path) or suite.startswith('.') or suite.startswith('_'):

• Read up
  Read up

Function from_grouped_postings has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    def from_grouped_postings(cls, grouped_postings: _GroupedPostings) -> '_ConversionTable':
        entries_by_currency = collections.defaultdict[str, set[_UnambiguousConversionTableEntry]](set)
        no_conversion_currencies: set[str] = set()
        weighted: list[_PostingPolicy] = grouped_postings.weighted
        prorated: list[_PostingPolicy] = grouped_postings.prorated

• Read up
  Read up

Function deduplicate_open_close has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def deduplicate_open_close(entries: list[Directive]) -> list[Directive]:
    """Deduplicates Open / Close directives.
    
    This deduplicates Open / Close directives in a simple way because duplication inside each file should have been
    caught during loading.

• Read up
  Read up

Function _override_policy_def has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def _override_policy_def(
        policy_def: PolicyDefinition,
        parent: Optional[PolicyDefinition],
        *,
        is_ephemeral: bool,

• Read up
  Read up

Function merge_narration has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def merge_narration(entry: Directive, comment_narrations: dict[str, dict[int, str]]) -> Directive:
    if not isinstance(entry, Transaction):
        return entry
    narrations = []
    for posting in entry.postings:

• Read up
  Read up

Function subgraphs has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def subgraphs(self) -> Iterable[set[_Node]]:
        visited = set()
        for new_node in self._new_nodes:
            if new_node in visited:
                continue

• Read up
  Read up

Function test_ok has 28 lines of code (exceeds 25 allowed). Consider refactoring.

Open

def test_ok() -> None:
    entries, errors = load(_FOO_TEXT)
    assert not errors
    txn = entries[-15]
    assert isinstance(txn, Transaction)

Function is_small has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def is_small(self, tolerances: decimal.Decimal | dict[str, decimal.Decimal]) -> bool:
        if isinstance(tolerances, decimal.Decimal):
            for balance in self._positions.values():
                if abs(balance) > tolerances:
                    return False

• Read up
  Read up

Function get_balance_policy has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def get_balance_policy(self, balance: Balance) -> Optional[Policy[WeightedOwnership]]:
        has_explicit_policy = False
        balance_policy_def = try_parse_policy_definition(balance.meta)
        if balance_policy_def:
            balance_policy_def = self._resolve_parent(balance_policy_def)

• Read up
  Read up

Function assert_same_errors has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

def assert_same_errors(actuals: list[error_lib.Error], expecteds: ExpectedErrors) -> None:
    actual_errors = ExpectedErrors()
    matched_expected_errors = ExpectedErrors()
    unmatched_expected_errors = copy.deepcopy(expecteds)
    for actual in actuals:

• Read up
  Read up