[Fix] Splitted transactions do not have category

Adds uncategorized method to the DB client to retrieve transactions w/o
a category AND not splitted.
This commit is contained in:
Luís Murta 2023-01-29 23:39:46 +00:00
parent abff76ad4e
commit f943374ade
Signed by: satprog
GPG Key ID: 169EF1BBD7049F94
2 changed files with 19 additions and 1 deletions

View File

@ -16,7 +16,7 @@ def interactive(manager: Manager):
tags = session.get(type.Tag) tags = session.get(type.Tag)
print(f"Available tags: {[t.name for t in tags]}") print(f"Available tags: {[t.name for t in tags]}")
transactions = session.get(type.Transaction, ~type.Transaction.category.has()) transactions = session.uncategorized()
print(f"{len(transactions)} transactions left to categorize") print(f"{len(transactions)} transactions left to categorize")
for transaction in sorted(transactions): for transaction in sorted(transactions):

View File

@ -2,6 +2,7 @@ from dataclasses import asdict
from sqlalchemy import create_engine, delete, select, update from sqlalchemy import create_engine, delete, select, update
from sqlalchemy.dialects.postgresql import insert from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from sqlalchemy.sql.expression import false
from typing import Sequence, Type, TypeVar from typing import Sequence, Type, TypeVar
from pfbudget.db.model import ( from pfbudget.db.model import (
@ -9,6 +10,7 @@ from pfbudget.db.model import (
CategoryGroup, CategoryGroup,
CategorySchedule, CategorySchedule,
Link, Link,
Transaction,
) )
@ -57,6 +59,22 @@ class DbClient:
return self.__session.scalars(stmt).all() return self.__session.scalars(stmt).all()
def uncategorized(self) -> Sequence[Transaction]:
"""Selects all valid uncategorized transactions
At this moment that includes:
- Categories w/o category
- AND non-split categories
Returns:
Sequence[Transaction]: transactions left uncategorized
"""
stmt = (
select(Transaction)
.where(~Transaction.category.has())
.where(Transaction.split == false())
)
return self.__session.scalars(stmt).all()
def add(self, rows: list): def add(self, rows: list):
self.__session.add_all(rows) self.__session.add_all(rows)