diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 02fc58e..82fdb5e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -25,3 +25,11 @@ repos: language: system require_serial: true args: ['--filter-files'] + - id: mypy + name: mypy + entry: mypy + types: [python] + language: system + require_serial: true + pass_filenames: false + args: ['financeager'] diff --git a/README.md b/README.md index a35221c..974a5a4 100644 --- a/README.md +++ b/README.md @@ -316,6 +316,62 @@ Please adhere to test-driven development, if possible: When adding a feature, or If you added a non-cosmetic change (i.e. a change in functionality, e.g. a bug fix or a new feature), please update `Changelog.md` accordingly as well. Check this README whether the content is still up to date. +### Type Checking + +This project uses type annotations throughout the codebase to improve code safety and developer experience. We use [mypy](https://mypy-lang.org/) for static type checking. + +#### Running Type Checks + +Type checking is integrated into the development workflow: + +```bash +# Run mypy manually +mypy financeager + +# Type checks are also run automatically via: +# - pre-commit hooks (before each commit) +# - CI pipeline (on pull requests and pushes) +``` + +#### Type Annotation Guidelines + +When contributing to the codebase, please follow these typing practices: + +- **All new functions and methods should have type annotations** for parameters and return values +- Use **modern Python 3.10+ type syntax** (e.g., `list[str]` instead of `List[str]`, `str | None` instead of `Optional[str]`) +- Import types from `typing` when needed: `from typing import Any, Dict, List, Optional` +- For complex cases, `# type: ignore` comments are acceptable with specific error codes +- Abstract methods and complex external library integrations may use gradual typing with `# type: ignore` + +#### Type Configuration + +The mypy configuration in `pyproject.toml` uses a gradual typing approach: +- Basic type safety is enforced +- Some strict checks are disabled during the migration period +- External dependencies without type stubs are ignored + +#### Migration to Strict Typing + +To progressively tighten the typing configuration: + +1. **Enable stricter checks gradually** by uncommenting options in `pyproject.toml`: + ```toml + disallow_untyped_defs = true # Require all functions to have type annotations + disallow_incomplete_defs = true # Disallow partially typed function definitions + disallow_untyped_decorators = true # Require decorators to be typed + ``` + +2. **Re-enable unused ignore warnings** once type annotations are complete: + ```toml + warn_unused_ignores = true + ``` + +3. **Remove type: ignore comments** as you fix the underlying type issues + +4. **Update to use more specific types** instead of `Any` where possible + +For more information about Python typing, see the [official documentation](https://docs.python.org/3/library/typing.html). + ## Releasing 1. Tag the latest commit on master by incrementing the current version accordingly (scheme `vmajor.minor.patch`). diff --git a/financeager/__init__.py b/financeager/__init__.py index 56f07a8..bbd6abc 100644 --- a/financeager/__init__.py +++ b/financeager/__init__.py @@ -1,6 +1,6 @@ import os.path from importlib.metadata import version -from logging import DEBUG, WARN, Formatter, StreamHandler, getLogger, handlers +from logging import DEBUG, WARN, Formatter, Logger, StreamHandler, getLogger, handlers import platformdirs @@ -47,14 +47,14 @@ FORMATTER = Formatter(fmt="%(levelname)s %(asctime)s %(name)s:%(lineno)d %(message)s") -def init_logger(name): +def init_logger(name: str) -> Logger: """Set up module logger. Library loggers are assigned the package logger as parent. Any records are propagated to the parent package logger. """ logger = getLogger(name) logger.setLevel(DEBUG) - if logger.parent.name == "root": + if logger.parent is not None and logger.parent.name == "root": # Library logger; probably has NullHandler logger.parent = LOGGER @@ -64,19 +64,19 @@ def init_logger(name): return logger -def setup_log_file_handler(log_dir=LOG_DIR): +def setup_log_file_handler(log_dir: str = LOG_DIR) -> None: """Create RotatingFileHandler for package logger, storing logs in 'log_dir' (default: LOG_DIR). The directory is created if not existing. """ os.makedirs(log_dir, exist_ok=True) file_handler = handlers.RotatingFileHandler( - os.path.join(log_dir, "log"), maxBytes=5e6, backupCount=5 + os.path.join(log_dir, "log"), maxBytes=int(5e6), backupCount=5 ) file_handler.setFormatter(FORMATTER) LOGGER.addHandler(file_handler) -def make_log_stream_handler_verbose(): +def make_log_stream_handler_verbose() -> None: """Make handler show debug messages using more informative format.""" _stream_handler.setLevel(DEBUG) _stream_handler.setFormatter(FORMATTER) diff --git a/financeager/cli.py b/financeager/cli.py index fd46e9b..4e83584 100644 --- a/financeager/cli.py +++ b/financeager/cli.py @@ -7,6 +7,7 @@ import time from datetime import datetime from importlib.metadata import entry_points +from typing import Any import argcomplete from dateutil import parser as du_parser @@ -92,7 +93,7 @@ def run(command, configuration, plugins=None, verbose=False, sinks=None, **param if verbose: make_log_stream_handler_verbose() - formatting_options = {} + formatting_options: dict[str, Any] = {} def _info(message): """Wrapper to format message and propagate it to stdout. The original @@ -296,9 +297,11 @@ def _parse_command(args=None, plugins=None): add_parser.add_argument("name", help="entry name") add_parser.add_argument("value", type=float, help="entry value") - add_parser.add_argument( + category_add_arg = add_parser.add_argument( "-c", "--category", default=None, help="entry category" - ).completer = argcomplete.ChoicesCompleter(categories) + ) + cat_arg = category_add_arg + cat_arg.completer = argcomplete.ChoicesCompleter(categories) # type: ignore add_parser.add_argument("-d", "--date", default=None, help="entry date") add_parser.add_argument( @@ -360,9 +363,8 @@ def _parse_command(args=None, plugins=None): ) update_parser.add_argument("-n", "--name", help="new name") update_parser.add_argument("-v", "--value", type=float, help="new value") - update_parser.add_argument("-c", "--category", help="new category").completer = ( - argcomplete.ChoicesCompleter(categories) - ) + category_arg = update_parser.add_argument("-c", "--category", help="new category") + category_arg.completer = argcomplete.ChoicesCompleter(categories) # type: ignore update_parser.add_argument( "-d", "--date", help="new date (for standard entries only)" ) @@ -507,7 +509,7 @@ def _parse_command(args=None, plugins=None): ]: subparser.add_argument( "-p", "--pocket", help="name of pocket to modify or query" - ).completer = argcomplete.ChoicesCompleter( + ).completer = argcomplete.ChoicesCompleter( # type: ignore[attr-defined] pocket_names(financeager.DATA_DIR) ) diff --git a/financeager/clients.py b/financeager/clients.py index 7d8ccf0..68110d0 100644 --- a/financeager/clients.py +++ b/financeager/clients.py @@ -3,21 +3,25 @@ import os.path import traceback from collections import namedtuple +from typing import Any import financeager from . import exceptions, init_logger, localserver, plugin +from .config import Configuration logger = init_logger(__name__) -def create(*, configuration, sinks, plugins): +def create( + *, configuration: Configuration, sinks: Any, plugins: list[plugin.PluginBase] | None +) -> "Client": """Factory to create the Client subclass suitable to the given configuration. Clients of service plugins are taken into account if specified. The sinks are passed into the Client. """ - clients = { + clients: dict[str, type[Client]] = { "local": LocalServerClient, } @@ -40,16 +44,16 @@ class Client: Sinks = namedtuple("Sinks", ["info", "error"]) - def __init__(self, *, configuration, sinks): + def __init__(self, *, configuration: Configuration, sinks: "Client.Sinks") -> None: """Store the specified configuration and sinks as attributes. The subclass implementation must set up the proxy. """ - self.proxy = None + self.proxy: Any = None self.configuration = configuration self.sinks = sinks - self.latest_exception = None + self.latest_exception: Exception | None = None - def safely_run(self, command, **params): + def safely_run(self, command: str, **params: Any) -> bool: """Execute self.proxy.run() while handling any errors. A caught exception is stored in the 'latest_exception' attribute. Return whether execution was successful. @@ -76,20 +80,20 @@ def safely_run(self, command, **params): return success - def shutdown(self): + def shutdown(self) -> None: """Routine to run at the end of the Client lifecycle.""" class LocalServerClient(Client): """Client for communicating with the financeager localserver.""" - def __init__(self, *, configuration, sinks): + def __init__(self, *, configuration: Configuration, sinks: "Client.Sinks") -> None: """Set up proxy.""" super().__init__(configuration=configuration, sinks=sinks) self.proxy = localserver.Proxy(data_dir=financeager.DATA_DIR) - def safely_run(self, command, **params): + def safely_run(self, command: str, **params: Any) -> bool: """Run the parent method, and for certain modifying commands, fetch category names from the server and store them in the cache. """ @@ -103,7 +107,7 @@ def safely_run(self, command, **params): logger.debug(str(e)) return success - def _write_categories_for_cli_completion(self): + def _write_categories_for_cli_completion(self) -> None: # There might be different categories for each pocket. However when reading the # cache at the time of building the CLI completion, the target pocket cannot be # determined. It's assumed the default pocket is most relevant, hence its @@ -114,6 +118,6 @@ def _write_categories_for_cli_completion(self): # The category cache is a line-separated list of names f.write("\n".join(categories)) - def shutdown(self): + def shutdown(self) -> None: """Instruct stopping of Server.""" self.proxy.run("stop") diff --git a/financeager/config.py b/financeager/config.py index bd898f3..9115c67 100644 --- a/financeager/config.py +++ b/financeager/config.py @@ -1,6 +1,7 @@ """Configuration of the financeager application.""" from configparser import ConfigParser, NoOptionError, NoSectionError +from typing import Any from financeager import plugin @@ -16,7 +17,11 @@ class Configuration: configuration can be customized by settings specified in a config file. """ - def __init__(self, filepath=None, plugins=None): + def __init__( + self, + filepath: str | None = None, + plugins: list[plugin.PluginBase] | None = None, + ) -> None: """Initialize the default configuration, overwrite with custom configuration from file if available, and eventually validate the loaded configuration. @@ -30,7 +35,7 @@ def __init__(self, filepath=None, plugins=None): self._plugins = plugins or [] self._parser = ConfigParser() - self._option_types = {} + self._option_types: dict[str, dict[str, str]] = {} self._init_defaults() self._init_option_types() @@ -38,7 +43,7 @@ def __init__(self, filepath=None, plugins=None): self._load_custom_config() self._validate() - def _init_defaults(self): + def _init_defaults(self) -> None: self._parser["SERVICE"] = { "name": "local", } @@ -49,11 +54,11 @@ def _init_defaults(self): for p in self._plugins: p.config.init_defaults(self._parser) - def _init_option_types(self): + def _init_option_types(self) -> None: for p in self._plugins: p.config.init_option_types(self._option_types) - def _load_custom_config(self): + def _load_custom_config(self) -> None: """Update config values according to customization in config file.""" if self._filepath is None: return @@ -78,13 +83,13 @@ def _load_custom_config(self): continue self._parser[section][item] = custom_value - def get_section(self, section): + def get_section(self, section: str) -> dict[str, Any]: """Return a dictionary of options of the requested section. If an option is typed, a converted value is returned. """ return {o: self.get_option(section, o) for o in self._parser.options(section)} - def get_option(self, section, option): + def get_option(self, section: str, option: str) -> Any: """Return the requested option of the configuration. If an option is typed, a converted value is returned. """ @@ -94,14 +99,14 @@ def get_option(self, section, option): # Option type not specified, assuming str option_type = None - if option_type in ("int", "float", "boolean"): + if option_type is not None and option_type in ("int", "float", "boolean"): get = getattr(self._parser, f"get{option_type}") else: get = self._parser.get return get(section, option) - def _validate(self): + def _validate(self) -> None: """Validate certain options of the configuration. Typed options are validated for possible conversion. diff --git a/financeager/entries.py b/financeager/entries.py index 48107f4..8edb74b 100644 --- a/financeager/entries.py +++ b/financeager/entries.py @@ -2,6 +2,7 @@ query results.""" import time +from typing import Any from . import POCKET_DATE_FORMAT @@ -12,7 +13,7 @@ class Entry: The name field is stored in lowercase, simplifying searching from the parent listing. The value is rendered absolute to simplify sorting.""" - def __init__(self, name, value): + def __init__(self, name: str, value: float | int) -> None: """:type name: str :type value: float or int """ @@ -26,7 +27,9 @@ class BaseEntry(Entry): DATE_FORMAT = "%y-%m-%d" - def __init__(self, name, value, date, eid=0): + def __init__( + self, name: str, value: float | int, date: str, eid: int | str = 0 + ) -> None: """:type eid: int or string, will be converted to int :type date: str of valid format """ @@ -43,22 +46,24 @@ class CategoryEntry(Entry): DEFAULT_NAME = "unspecified" - def __init__(self, name, entries=None): + def __init__( + self, name: str | None, entries: list[BaseEntry] | None = None + ) -> None: """:type entries: list[BaseEntry]""" super().__init__(name=name or self.DEFAULT_NAME, value=0.0) - self.entries = [] + self.entries: list[BaseEntry] = [] if entries is not None: for base_entry in entries: self.append(base_entry) - def append(self, base_entry): + def append(self, base_entry: BaseEntry) -> None: """Append a BaseEntry to the category and update the value.""" self.entries.append(base_entry) self.value += base_entry.value -def prettify(element, *, default_category): +def prettify(element: dict[str, Any], *, default_category: str) -> str: """Return element properties formatted as list. The type of the element (recurrent or standard) is inferred by the presence of the 'frequency' property. If the element's 'category' property is None, use 'default_category'. @@ -70,10 +75,17 @@ def prettify(element, *, default_category): # Define order of listed properties if recurrent: - properties = ("name", "value", "frequency", "start", "end", "category") + properties: list[str] = [ + "name", + "value", + "frequency", + "start", + "end", + "category", + ] longest_property_length = 9 # frequency else: - properties = ("name", "value", "date", "category") + properties = ["name", "value", "date", "category"] longest_property_length = 8 # category if element["category"] is None: diff --git a/financeager/listing.py b/financeager/listing.py index 02c8259..aec82f0 100644 --- a/financeager/listing.py +++ b/financeager/listing.py @@ -1,6 +1,7 @@ """Tabular, frontend-representation of financeager pocket.""" from json import dumps as jdumps +from typing import Any, Generator from . import DEFAULT_TABLE, RECURRENT_TABLE from .entries import BaseEntry, CategoryEntry @@ -12,12 +13,19 @@ class Listing: CategoryEntries, second-level children are BaseEntries. Generator methods are provided to iterate over these.""" - def __init__(self, name=None, categories=None): + def __init__( + self, name: str | None = None, categories: list[CategoryEntry] | None = None + ) -> None: self.name = name or "Listing" self.categories = categories or [] @classmethod - def from_elements(cls, elements, default_category=None, name=None): + def from_elements( + cls, + elements: list[dict[str, Any]], + default_category: str | None = None, + name: str | None = None, + ) -> "Listing": """Create listing from list of element dictionaries""" listing = cls(name=name) for element in elements: @@ -25,7 +33,9 @@ def from_elements(cls, elements, default_category=None, name=None): listing.add_entry(BaseEntry(**element), category_name=category) return listing - def add_entry(self, entry, category_name=None): + def add_entry( + self, entry: CategoryEntry | BaseEntry, category_name: str | None = None + ) -> None: """Add a Category- or BaseEntry to the listing. Category names are unique, i.e. a CategoryEntry is discarded if one with identical name (case INsensitive) already exists. @@ -44,7 +54,7 @@ def add_entry(self, entry, category_name=None): else: raise TypeError(f"Invalid entry type: {entry}") - def category_fields(self, field_type): + def category_fields(self, field_type: str) -> Generator[Any, None, None]: """Generator iterating over the field specified by `field_type` of the first-level children (CategoryEntries) of the listing. @@ -57,19 +67,19 @@ def category_fields(self, field_type): yield getattr(category_entry, field_type) @property - def category_entry_names(self): + def category_entry_names(self) -> Generator[str, None, None]: """Convenience generator method yielding category names.""" for category_name in self.category_fields("name"): yield category_name - def _get_category_entry(self, category_name): + def _get_category_entry(self, category_name: str | None) -> CategoryEntry: """Fetch CategoryEntry searching for given `category_name` or return a new instance that is automatically added to the Listing's categories. The search is case insensitive. :return: CategoryEntry """ - category_name = category_name.lower() + category_name = (category_name or CategoryEntry.DEFAULT_NAME).lower() for category_entry in self.categories: if category_entry.name == category_name: @@ -80,18 +90,18 @@ def _get_category_entry(self, category_name): self.add_entry(category_entry) return category_entry - def total_value(self): + def total_value(self) -> float: """Return total value of the listing.""" - return sum(v for v in self.category_fields("value")) + return sum(v for v in self.category_fields("value")) # type: ignore def prettify( - elements, - recurrent_only=False, - json=False, - default_category=None, - **listing_options, -): + elements: dict[str, Any], + recurrent_only: bool = False, + json: bool = False, + default_category: str | None = None, + **listing_options: Any, +) -> str: """Sort the given elements (type acc. to Pocket._search_all_tables) by positive and negative value and print tabular representation. @@ -104,18 +114,21 @@ def prettify( return jdumps(elements) if recurrent_only: - entry_sort = listing_options.get("entry_sort") - return richify_recurrent_elements(elements, entry_sort=entry_sort) + sort = listing_options.get("entry_sort") + res = richify_recurrent_elements(elements, entry_sort=sort) # type: ignore + return res # type: ignore listings = _derive_listings(elements, default_category=default_category) - return richify_listings(listings, **listing_options) + return richify_listings(listings, **listing_options) # type: ignore[return-value] -def _derive_listings(elements, *, default_category): - earnings = [] - expenses = [] +def _derive_listings( + elements: dict[str, Any], *, default_category: str | None +) -> list[Listing] | None: + earnings: list[dict[str, Any]] = [] + expenses: list[dict[str, Any]] = [] - def _sort(eid, element): + def _sort(eid: int, element: dict[str, Any]) -> None: # Copying avoids modifying the original element. Flattening is in order # to distinguish recurrent entries (they have the same element ID which # thus can't be used as dict key) @@ -136,7 +149,7 @@ def _sort(eid, element): _sort(eid, element) if not earnings and not expenses: - return + return None listing_earnings = Listing.from_elements( earnings, default_category=default_category, name="Earnings" diff --git a/financeager/localserver.py b/financeager/localserver.py index 45c3f2c..190a432 100644 --- a/financeager/localserver.py +++ b/financeager/localserver.py @@ -1,6 +1,8 @@ """Local server proxy for direct communication (client and server reside in common process).""" +from typing import Any + from . import exceptions, init_logger, server logger = init_logger(__name__) @@ -9,7 +11,7 @@ class Proxy(server.Server): """Subclass mocking a locally running server. Convenient for testing""" - def run(self, command, **kwargs): + def run(self, command: str, **kwargs: Any) -> dict[str, Any]: """Run command on local server. Exceptions are propagated upwards. Call run('stop') as last operation to properly close the databases before exiting the process. diff --git a/financeager/plugin.py b/financeager/plugin.py index 095cdbf..70d9b9d 100644 --- a/financeager/plugin.py +++ b/financeager/plugin.py @@ -1,6 +1,9 @@ """Support for plugin development.""" import abc +from argparse import ArgumentParser +from configparser import ConfigParser +from typing import Any class PluginConfiguration(abc.ABC): @@ -9,19 +12,19 @@ class PluginConfiguration(abc.ABC): class.""" @abc.abstractmethod - def init_defaults(self, config_parser): + def init_defaults(self, config_parser: ConfigParser) -> None: """Initialize configuration defaults by extending the given ConfigParser by relevant sections. """ - def init_option_types(self, option_types): + def init_option_types(self, option_types: dict[str, dict[str, str]]) -> None: """Specify types of plugin options (int, float, boolean) if conversion at the time of retrieving the option is desired. The given dictionary should be modified in-place by adding the type to the corresponding section and option. """ - def validate(self, config): + def validate(self, config: Any) -> None: """Validate content of the Configuration object specific to the plugin. Typed options are implicitly validated for conversion by the Configuration object already. @@ -34,7 +37,7 @@ class PluginCliOptions(abc.ABC): this class.""" @abc.abstractmethod - def extend(self, command_parser): + def extend(self, command_parser: ArgumentParser) -> None: """Extend the 'command' subparser of the financeager CLI by additional commands and/or arguments. Remember that the corresponding Client class must provide an @@ -43,12 +46,18 @@ def extend(self, command_parser): class DefaultPluginCliOptions(PluginCliOptions): - def extend(self, _): + def extend(self, _: ArgumentParser) -> None: pass class PluginBase: - def __init__(self, *, name, config, cli_options=None): + def __init__( + self, + *, + name: str, + config: PluginConfiguration, + cli_options: PluginCliOptions | None = None, + ) -> None: """Set plugin name, config (a PluginConfiguration instance), and optional CLI options (a PluginCliOptions instance).""" self.name = name @@ -62,6 +71,13 @@ class ServicePlugin(PluginBase): communication with the service. """ - def __init__(self, *, name, config, client, cli_options=None): + def __init__( + self, + *, + name: str, + config: PluginConfiguration, + client: Any, + cli_options: PluginCliOptions | None = None, + ) -> None: super().__init__(name=name, config=config, cli_options=cli_options) self.client = client diff --git a/financeager/pocket.py b/financeager/pocket.py index 49865ae..3efd603 100644 --- a/financeager/pocket.py +++ b/financeager/pocket.py @@ -4,6 +4,7 @@ from abc import ABC, abstractmethod from collections import Counter, defaultdict from datetime import datetime as dt +from typing import Any from dateutil import rrule from marshmallow import Schema, ValidationError, fields, validate @@ -19,7 +20,7 @@ exceptions, ) -_DEFAULT_CATEGORY = None +_DEFAULT_CATEGORY: None = None FREQUENCY_CHOICES = [ "yearly", "half-yearly", @@ -54,18 +55,18 @@ class RecurrentEntrySchema(EntryBaseSchema): class Pocket(ABC): - def __init__(self, name=None): + def __init__(self, name: str | None = None) -> None: """Create Pocket object. Its name defaults to the current year if not specified. """ self._name = f"{name or DEFAULT_POCKET_NAME}" @property - def name(self): + def name(self) -> str: return self._name @abstractmethod - def add_entry(self, table_name=None, **kwargs): + def add_entry(self, table_name: str | None = None, **kwargs: Any) -> int: """Add an entry (standard or recurrent) to the database. If 'table_name' is not specified, the kwargs name, value[, category, @@ -104,7 +105,9 @@ def add_entry(self, table_name=None, **kwargs): """ @abstractmethod - def get_entry(self, eid, table_name=None): + def get_entry( + self, eid: int | str, table_name: str | None = None + ) -> dict[str, Any]: """Get entry specified by eid in the table table_name. :param eid: entry ID @@ -117,7 +120,9 @@ def get_entry(self, eid, table_name=None): """ @abstractmethod - def update_entry(self, eid, table_name=None, **kwargs): + def update_entry( + self, eid: int | str, table_name: str | None = None, **kwargs: Any + ) -> int: """Update one or more fields of a single entry. :param eid: entry ID of the entry to be updated @@ -130,7 +135,7 @@ def update_entry(self, eid, table_name=None, **kwargs): """ @abstractmethod - def remove_entry(self, eid, table_name=None): + def remove_entry(self, eid: int | str, table_name: str | None = None) -> int: """Remove an entry from the database given its ID. :param eid: ID of the element to be deleted. @@ -144,7 +149,9 @@ def remove_entry(self, eid, table_name=None): """ @abstractmethod - def get_entries(self, filters=None, recurrent_only=False): + def get_entries( + self, filters: dict[str, Any] | None = None, recurrent_only: bool = False + ) -> dict[str, Any]: """Get entries that match the items of the filters dict, if specified. If `recurrent_only` is true, return a list of all entries of the @@ -156,16 +163,18 @@ def get_entries(self, filters=None, recurrent_only=False): """ @abstractmethod - def get_categories(self): + def get_categories(self) -> list[str]: """Return unique category names in alphabetical order.""" @abstractmethod - def close(self): + def close(self) -> None: """Close underlying database.""" class TinyDbPocket(Pocket): - def __init__(self, name=None, data_dir=None, **kwargs): + def __init__( + self, name: str | None = None, data_dir: str | None = None, **kwargs: Any + ) -> None: """Create a pocket with a TinyDB database backend, identified by 'name'. If 'data_dir' is given, the database storage type is JSON (the storage filepath is derived from the Pocket's name). Otherwise the data is @@ -179,7 +188,7 @@ def __init__(self, name=None, data_dir=None, **kwargs): # evaluate args/kwargs for TinyDB constructor. This overwrites the # 'storage' kwarg if explicitly passed if data_dir is None: - args = [] + args: list[Any] = [] kwargs["storage"] = storages.MemoryStorage else: args = [os.path.join(data_dir, f"{self.name}.json")] @@ -188,16 +197,21 @@ def __init__(self, name=None, data_dir=None, **kwargs): self._db = TinyDB(*args, **kwargs) self._create_category_cache() - def _create_category_cache(self): + def _create_category_cache(self) -> None: """The category cache assigns a counter for each element name in the database (excluding recurrent elements), keeping track of the categories the element was labeled with. This allows deriving the category of an element if not explicitly given.""" - self._category_cache = defaultdict(Counter) + self._category_cache: defaultdict[str, Counter[str]] = defaultdict(Counter) for element in self._db.all(): self._category_cache[element["name"]].update([element["category"]]) - def _preprocess_entry(self, raw_data=None, table_name=None, partial=False): + def _preprocess_entry( + self, + raw_data: dict[str, Any] | None = None, + table_name: str | None = None, + partial: bool = False, + ) -> dict[str, Any]: """Perform preprocessing steps (validation, conversion, substitution) of raw entry fields prior to adding it to the database. @@ -216,6 +230,9 @@ def _preprocess_entry(self, raw_data=None, table_name=None, partial=False): f"Unknown table name: {table_name}" ) + if raw_data is None: + raw_data = {} + self._remove_redundant_fields(table_name, raw_data) validated_fields = self._validate_entry( @@ -231,7 +248,7 @@ def _preprocess_entry(self, raw_data=None, table_name=None, partial=False): return converted_fields @staticmethod - def _remove_redundant_fields(table_name, raw_data): + def _remove_redundant_fields(table_name: str, raw_data: dict[str, Any]) -> None: """The raw data (e.g. parsed from the command line) might contain fields that are not required by the given table type and hence, they crash the schematics validation ('Rogue field' error). This method removes @@ -247,7 +264,9 @@ def _remove_redundant_fields(table_name, raw_data): raw_data.pop(field, None) @staticmethod - def _validate_entry(raw_data, table_name, **schema_kwargs): + def _validate_entry( + raw_data: dict[str, Any], table_name: str, **schema_kwargs: Any + ) -> dict[str, Any]: """Validate raw entry data acc. to ValidationSchema. :return: primitive (type-correct) representation of fields @@ -263,18 +282,18 @@ def _validate_entry(raw_data, table_name, **schema_kwargs): try: schema = ValidationSchema(**schema_kwargs) validated_data = schema.load(raw_data) - return schema.dump(validated_data) + return schema.dump(validated_data) # type: ignore[no-any-return] except ValidationError as e: infos = [ f"{field}: {'; '.join(messages)}" - for field, messages in e.messages.items() + for field, messages in e.messages.items() # type: ignore[union-attr] ] raise exceptions.PocketValidationFailure( "Invalid input data:\n{}".format("\n".join(infos)) ) @staticmethod - def _convert_fields(**fields): + def _convert_fields(**fields: Any) -> dict[str, Any]: """Convert string field values to lowercase for storage. Fields with value None are discarded. """ @@ -291,7 +310,7 @@ def _convert_fields(**fields): return converted_fields - def _substitute_none_fields(self, table_name, **fields): + def _substitute_none_fields(self, table_name: str, **fields: Any) -> dict[str, Any]: """Substitute optional fields by defaults.""" substituted_fields = fields.copy() @@ -442,7 +461,10 @@ def _search_all_tables(self, condition): :return: dict """ - elements = {DEFAULT_TABLE: {}, RECURRENT_TABLE: defaultdict(list)} + elements: dict[str, dict[int, Any] | defaultdict[int, list[Any]]] = { + DEFAULT_TABLE: {}, + RECURRENT_TABLE: defaultdict(list), + } for element in self._db.search(condition): elements[DEFAULT_TABLE][element.doc_id] = element @@ -501,7 +523,7 @@ def _create_recurrent_elements(self, element): name = f"{name}, day {date.strftime('%-j').lower()}" yield Document( - doc_id=None, + doc_id=None, # type: ignore[arg-type] value=dict( name=name, value=element["value"], diff --git a/financeager/rich.py b/financeager/rich.py index 5b8c42d..192c9af 100644 --- a/financeager/rich.py +++ b/financeager/rich.py @@ -1,17 +1,23 @@ +from typing import TYPE_CHECKING, Any + from rich import box from rich.panel import Panel from rich.table import Table +from rich.table import Table as RichTable from . import DEFAULT_BASE_ENTRY_SORT_KEY, DEFAULT_CATEGORY_ENTRY_SORT_KEY +if TYPE_CHECKING: + from .listing import Listing + def richify_listings( - listings, - stacked_layout=False, - category_sort=None, - category_percentage=False, - entry_sort=None, -): + listings: list["Listing"] | None, + stacked_layout: bool = False, + category_sort: str | None = None, + category_percentage: bool = False, + entry_sort: str | None = None, +) -> str | RichTable: """Create and return rich.Table from listings acc. to given options. :param stacked_layout: If True, listings are displayed one by one :param category_sort: Field governing category sorting (name, value) @@ -22,18 +28,19 @@ def richify_listings( if not listings: return "No entries found." - tables = [] + tables: list[RichTable] = [] category_sort = category_sort or DEFAULT_CATEGORY_ENTRY_SORT_KEY entry_sort = entry_sort or DEFAULT_BASE_ENTRY_SORT_KEY totals = [ls.total_value() for ls in listings] # Calculate maximum column widths for stacked layout alignment - max_widths = [None, None, None, None] + max_widths: list[int | None] = [None, None, None, None] if stacked_layout: - max_widths = _calculate_max_column_widths( + max_widths_calculated = _calculate_max_column_widths( listings, totals, category_percentage, category_sort, entry_sort ) + max_widths = [x for x in max_widths_calculated] # Build tables for listings for listing, total in zip(listings, totals): @@ -79,7 +86,7 @@ def richify_listings( str(entry.eid), ) - def nr_rows(listing): + def nr_rows(listing: "Listing") -> int: if category_percentage: return len(listing.categories) return len(listing.categories) + sum(len(c.entries) for c in listing.categories) @@ -115,7 +122,9 @@ def nr_rows(listing): return grid -def richify_recurrent_elements(elements, entry_sort=None): +def richify_recurrent_elements( + elements: list[dict[str, Any]], entry_sort: str | None = None +) -> RichTable: """Create and return rich.Table from recurrent elements acc. to given options. :param entry_sort: Field governing base entry sorting (name, value, ID, category, start, end, frequency) @@ -144,8 +153,12 @@ def richify_recurrent_elements(elements, entry_sort=None): def _calculate_max_column_widths( - listings, totals, category_percentage, category_sort, entry_sort -): + listings: list["Listing"], + totals: list[float], + category_percentage: bool, + category_sort: str, + entry_sort: str, +) -> list[int]: """Calculate maximum column widths needed across all listings for consistent alignment.""" max_name_width = 0 diff --git a/financeager/server.py b/financeager/server.py index e36efdb..59b6798 100644 --- a/financeager/server.py +++ b/financeager/server.py @@ -2,6 +2,7 @@ import glob import os.path +from typing import Any from . import DEFAULT_POCKET_NAME, exceptions, init_logger, pocket @@ -15,11 +16,11 @@ class Server: Kwargs (f.i. storage) are passed to the TinyDbPocket member. """ - def __init__(self, **kwargs): - self._pockets = {} + def __init__(self, **kwargs: Any) -> None: + self._pockets: dict[str, pocket.TinyDbPocket] = {} self._pocket_kwargs = kwargs - def run(self, command, **kwargs): + def run(self, command: str, **kwargs: Any) -> dict[str, Any]: """The requested pocket is created if not yet present. The method of `Pocket` corresponding to the given `command` is called. All `kwargs` are passed on. A json-like response is returned. @@ -63,7 +64,7 @@ def run(self, command, **kwargs): except exceptions.PocketException as e: return {"error": e} - def _get_pocket(self, name=None): + def _get_pocket(self, name: str | None = None) -> pocket.TinyDbPocket: """Get the Pocket identified by 'name' from the Pockets dictionary. If the Pocket does not exist, it is created and returned. If 'name' is None, the default pocket name is used as defined in __init__.py @@ -81,7 +82,7 @@ def _get_pocket(self, name=None): return pd - def _pocket_names(self): + def _pocket_names(self) -> list[str]: """Return names of pockets currently organized by the server. If persistent data storage was specified, all JSON files present in the 'data_dir' are also taken into account. @@ -96,7 +97,12 @@ def _pocket_names(self): names.update(pocket_names(data_dir)) return sorted(names) - def _copy_entry(self, source_pocket=None, destination_pocket=None, **kwargs): + def _copy_entry( + self, + source_pocket: str | None = None, + destination_pocket: str | None = None, + **kwargs: Any, + ) -> int: """Copy an entry (specified by ID and table_name) from the source pocket to the destination pocket. @@ -104,16 +110,16 @@ def _copy_entry(self, source_pocket=None, destination_pocket=None, **kwargs): :return: ID of copied entry :raises: PocketException if the source entry does not exist """ - source_pocket = self._get_pocket(source_pocket) - entry_to_copy = source_pocket.get_entry(**kwargs) + source_pocket_obj = self._get_pocket(source_pocket) + entry_to_copy = source_pocket_obj.get_entry(**kwargs) - destination_pocket = self._get_pocket(destination_pocket) - return destination_pocket.add_entry( + destination_pocket_obj = self._get_pocket(destination_pocket) + return destination_pocket_obj.add_entry( # type: ignore[no-any-return] table_name=kwargs.get("table_name"), **entry_to_copy ) -def pocket_names(data_dir): +def pocket_names(data_dir: str | None) -> set[str]: """Return names of all pockets (i.e. names of JSON files in the given data directory), or an empty set if the specified data directory is None. """ diff --git a/pyproject.toml b/pyproject.toml index b4d3a5a..9dfcb87 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,7 +52,9 @@ develop = [ "flake8-pyproject==1.2.3", "gitlint-core==0.19.1", 'isort==6.1.0', + 'mypy==1.17.1', 'pre-commit==4.3.0', + 'types-python-dateutil==2.9.0.20260508', ] packaging = [ "build", @@ -104,3 +106,20 @@ extend-ignore = [ "W503", # line break before binary operator "W504", # line break after binary operator ] + +[tool.mypy] +python_version = "3.10" +warn_return_any = true +warn_unused_configs = true +# These are disabled during gradual typing migration +# disallow_untyped_defs = true +# disallow_incomplete_defs = true +check_untyped_defs = true +# disallow_untyped_decorators = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = false +warn_no_return = true +warn_unreachable = true +strict_equality = true +extra_checks = true