Skip to content

Commit

Permalink
Merge pull request #5 from adnilsson/py310
Browse files Browse the repository at this point in the history
Refactor with Python 3.9 and 3.10 features
  • Loading branch information
adnilsson authored Oct 24, 2021
2 parents d0c0191 + 3954d9a commit da15652
Show file tree
Hide file tree
Showing 7 changed files with 61 additions and 55 deletions.
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ Also, I don't want to be forced to store all my bank data on their servers.

### Requirements

* Python 3.8+
* Python ^3.10

# User guide

Expand Down
46 changes: 25 additions & 21 deletions poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ authors = ["Adrian Nilsson"]
license = "MIT"

[tool.poetry.dependencies]
python = "^3.8"
python = "^3.10"
tomli = "^1.2.1"

[tool.poetry.dev-dependencies]
Expand Down
22 changes: 11 additions & 11 deletions src/config.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import enum
import tomli
from pathlib import Path
from typing import Any, Callable, Dict, Optional
from typing import Any, Callable


class TransactionFormat(enum.Flag):
Expand All @@ -25,14 +25,14 @@ def __init__(
name: str,
date_format: str,
date_column: str,
outflow_column: Optional[str]=None,
inflow_column: Optional[str]=None,
amount_column: Optional[str]=None,
payee_column: Optional[str]=None,
memo_column: Optional[str]=None,
category_column: Optional[str]=None,
csv_delimiter: Optional[str]=None,
normalizer: Optional[Callable[[str], str]]=None,
outflow_column: (str | None)=None,
inflow_column: (str | None)=None,
amount_column: (str | None)=None,
payee_column: (str | None)=None,
memo_column: (str | None)=None,
category_column: (str | None)=None,
csv_delimiter: (str | None)=None,
normalizer: (Callable[[str], str] | None)=None,
):
if date_column is None or date_column == '':
raise ValueError(f"The date column name is empty; {date_column=}")
Expand Down Expand Up @@ -125,10 +125,10 @@ def from_file(cls, toml_config: Path):
return cls.from_dict(config)

@classmethod
def from_dict(cls, toml_config: Dict[str, Any]):
def from_dict(cls, toml_config: dict[str, Any]):
name = toml_config['name']

csv_config: Dict[str, Any] = toml_config['csv']
csv_config: dict[str, Any] = toml_config['csv']
date_format = csv_config['date_format']
csv_delimiter = csv_config.get('delimiter')

Expand Down
37 changes: 21 additions & 16 deletions src/converter.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from datetime import datetime
from pathlib import Path
from typing import Dict, List, NamedTuple, Tuple
from typing import NamedTuple, TypeAlias
import csv
import warnings

Expand All @@ -27,6 +27,8 @@
#
# Any field can be left blank except the date

StrPair: TypeAlias = tuple[str, str]

class YnabHeader(NamedTuple):
""" Mapping to the column names specified by YNAB4
"""
Expand All @@ -48,7 +50,7 @@ def __init__(self, config: BankConfig):
self.parsedRows = []
self.numEmptyRows = 0

def convert(self, statement_csv: Path, toIgnore=None):
def convert(self, statement_csv: Path, toIgnore=None) -> bool:
toIgnore = [] if toIgnore is None else toIgnore

# Attempt to parse input file to a YNAB-formatted csv file
Expand All @@ -58,7 +60,7 @@ def convert(self, statement_csv: Path, toIgnore=None):

return self.writeOutput(parsed)

def readInput(self, statement_csv: Path, toIgnore) -> List[Dict[str, str]]:
def readInput(self, statement_csv: Path, toIgnore) -> list[dict[str, str]]:
with statement_csv.open(encoding='utf-8-sig', newline='') as f:
restkey='overflow'
reader = csv.DictReader(
Expand All @@ -67,15 +69,15 @@ def readInput(self, statement_csv: Path, toIgnore) -> List[Dict[str, str]]:
restkey=restkey,
skipinitialspace=True, # important since qouting won't work if there is leading whitespace
)
reader.fieldnames = [normalize(name) for name in reader.fieldnames]
reader.fieldnames = [self.config.normalizer(name) for name in reader.fieldnames]
try:
for raw_row in reader:
if (overflowing_columns := raw_row.get(restkey)):
msg = f"Exccess columns found: {overflowing_columns=}"
warnings.warn(msg, RuntimeWarning)
del raw_row[restkey]

row = {k: normalize(v) for k, v in raw_row.items()}
row = {k: self.config.normalizer(v) for k, v in raw_row.items()}
is_empty = all((len(v) == 0 for v in row.values()))
if is_empty:
warnings.warn(
Expand Down Expand Up @@ -113,7 +115,7 @@ def parseRows(self, bankRows):

return self.parsedRows

def _parseAmountField(self, bankline) -> Tuple[str, str]:
def _parseAmountField(self, bankline) -> StrPair:
amount = bankline[self.config.amount_column]
sign = '-' if amount[0] == '-' else '+'

Expand All @@ -122,7 +124,7 @@ def _parseAmountField(self, bankline) -> Tuple[str, str]:

return outflow, inflow

def _parseInflowOutflowFields(self, bankline) -> Tuple[str, str]:
def _parseInflowOutflowFields(self, bankline) -> StrPair:
outflow = bankline.get(self.config.outflow_column, '')
inflow = bankline.get(self.config.inflow_column, '')

Expand All @@ -131,12 +133,15 @@ def _parseInflowOutflowFields(self, bankline) -> Tuple[str, str]:

return outflow, inflow

def parseTransactionValue(self, bankline) -> Tuple[str, str]:
def parseTransactionValue(self, bankline) -> StrPair:
transactionParser = None
if self.config.transaction_format is TransactionFormat.AMOUNT:
transactionParser = self._parseAmountField
elif self.config.transaction_format is TransactionFormat.OUT_IN:
transactionParser = self._parseInflowOutflowFields
match self.config.transaction_format:
case TransactionFormat.AMOUNT:
transactionParser = self._parseAmountField
case TransactionFormat.OUT_IN:
transactionParser = self._parseInflowOutflowFields
case _:
raise RuntimeError(f"{self.config.transaction_format=} is not a valid TransactionFormat")

if transactionParser is None:
raise RuntimeError(f'expected {TransactionFormat.AMOUNT} or'
Expand All @@ -145,7 +150,7 @@ def parseTransactionValue(self, bankline) -> Tuple[str, str]:

return transactionParser(bankline)

def parseRow(self, bankline: Dict[str, str]):
def parseRow(self, bankline: dict[str, str]):
# must have outflow/inflow columns in YNAB4
outflow, inflow = self.parseTransactionValue(bankline)

Expand All @@ -164,10 +169,10 @@ def parseRow(self, bankline: Dict[str, str]):

return ynab_row

def writeOutput(self, parsedRows):
def writeOutput(self, parsedRows) -> bool:
hasWritten = False

if(parsedRows == None or len(parsedRows) == 0):
if parsedRows == None or len(parsedRows) == 0:
return hasWritten

with open('ynabImport.csv', 'w', encoding='utf-8', newline='') as outputFile:
Expand All @@ -181,7 +186,7 @@ def writeOutput(self, parsedRows):
finally:
return hasWritten

def namedtupleLen(tupleArg):
def namedtupleLen(tupleArg: NamedTuple) -> int:
return len(tupleArg._fields)

def readIgnore():
Expand Down
4 changes: 1 addition & 3 deletions tests/test_config.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,10 @@
from os import name
from pathlib import Path
from typing import List
from src.config import BankConfig

import pytest

@pytest.fixture
def bank_config_files() -> List[Path]:
def bank_config_files() -> list[Path]:
config_files = [c for c in Path('banks').iterdir() if c.suffix == '.toml']
if len(config_files) == 0:
raise FileNotFoundError("No TOML config files found in the 'banks/' directory")
Expand Down
3 changes: 1 addition & 2 deletions tests/util.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
from pathlib import Path
from typing import Optional


def examples_dir():
Expand All @@ -20,7 +19,7 @@ def find_dir(dirname: str) -> Path:

return dir_path

def _find_dir(dirname: str, dir: Path) -> Optional[Path]:
def _find_dir(dirname: str, dir: Path) -> Path | None:
dirs = (d for d in dir.iterdir() if d.is_dir())
for d in dirs:
if d.name == dirname:
Expand Down

0 comments on commit da15652

Please sign in to comment.