k1nq
d582f15681
- Implemented tests for import polling functionality, ensuring correct behavior for active and terminal import statuses. - Created tests for the login page, validating user interactions and error handling. - Added unit tests for money utility functions, covering parsing, formatting, and summation of monetary values. - Developed tests for query keys, ensuring proper scoping and invalidation roots. chore: setup testing environment and configuration - Added setup file for testing with Vitest and Testing Library. - Configured TypeScript and Vite for the frontend project. - Created Docker and Nginx configurations for local development and production environments. - Established a PostgreSQL initialization script for setting up application roles. docs: add README and environment configuration for infra - Included README for the infra directory, detailing local development setup and service descriptions. - Created an example environment configuration file for local development with placeholders for sensitive information.
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from app.categorization.models import CategorizationRule, IncomeSource, Payee, PayeeAlias
|
|
from app.errors import conflict, not_found
|
|
from app.ledger.models import Category
|
|
|
|
|
|
def resolve_category_name(category: Category) -> str:
|
|
"""Standard categories resolve to a translation key; overrides/customs are literal."""
|
|
if category.custom_name is not None:
|
|
return category.custom_name
|
|
if category.name_override is not None:
|
|
return category.name_override
|
|
if category.system_key is not None:
|
|
return f"category.{category.system_key}"
|
|
return "category.unnamed"
|
|
|
|
|
|
def get_income_source(db: Session, *, space_id, income_source_id) -> IncomeSource:
|
|
obj = db.query(IncomeSource).filter_by(id=income_source_id, space_id=space_id).one_or_none()
|
|
if obj is None:
|
|
raise not_found("INCOME_SOURCE_NOT_FOUND", "Income source not found")
|
|
return obj
|
|
|
|
|
|
def ensure_income_source(db: Session, *, space_id, name: str) -> IncomeSource:
|
|
obj = db.query(IncomeSource).filter_by(space_id=space_id, name=name).one_or_none()
|
|
if obj is None:
|
|
obj = IncomeSource(space_id=space_id, name=name)
|
|
db.add(obj)
|
|
db.flush()
|
|
return obj
|
|
|
|
|
|
def get_or_create_payee(db: Session, *, space_id, name: str, alias: str | None = None) -> Payee:
|
|
payee = db.query(Payee).filter_by(space_id=space_id, name=name).one_or_none()
|
|
if payee is None:
|
|
payee = Payee(space_id=space_id, name=name)
|
|
db.add(payee)
|
|
db.flush()
|
|
if alias:
|
|
add_alias(db, space_id=space_id, payee_id=payee.id, alias=alias)
|
|
return payee
|
|
|
|
|
|
def add_alias(db: Session, *, space_id, payee_id, alias: str) -> PayeeAlias:
|
|
existing = db.query(PayeeAlias).filter_by(space_id=space_id, alias=alias).one_or_none()
|
|
if existing is not None:
|
|
if existing.payee_id != payee_id:
|
|
raise conflict(
|
|
"PAYEE_ALIAS_TAKEN",
|
|
"Alias already mapped to another payee",
|
|
)
|
|
return existing
|
|
obj = PayeeAlias(space_id=space_id, payee_id=payee_id, alias=alias)
|
|
db.add(obj)
|
|
db.flush()
|
|
return obj
|
|
|
|
|
|
def find_payee_by_description(db: Session, *, space_id, description: str) -> Payee | None:
|
|
if not description:
|
|
return None
|
|
alias = (
|
|
db.query(PayeeAlias)
|
|
.filter(PayeeAlias.space_id == space_id, PayeeAlias.alias == description)
|
|
.one_or_none()
|
|
)
|
|
if alias is not None:
|
|
return db.get(Payee, alias.payee_id)
|
|
payee = db.query(Payee).filter_by(space_id=space_id, name=description).one_or_none()
|
|
return payee
|
|
|
|
|
|
def rule_category_for_payee(db: Session, *, space_id, payee_id) -> CategorizationRule | None:
|
|
return (
|
|
db.query(CategorizationRule).filter_by(space_id=space_id, payee_id=payee_id).one_or_none()
|
|
)
|
|
|
|
|
|
def set_payee_rule(db: Session, *, space_id, payee_id, category_id, user_id) -> CategorizationRule:
|
|
payee = db.query(Payee).filter_by(id=payee_id, space_id=space_id).one_or_none()
|
|
if payee is None:
|
|
raise not_found("PAYEE_NOT_FOUND", "Payee not found")
|
|
rule = rule_category_for_payee(db, space_id=space_id, payee_id=payee_id)
|
|
if rule is None:
|
|
rule = CategorizationRule(
|
|
space_id=space_id,
|
|
payee_id=payee_id,
|
|
category_id=category_id,
|
|
created_by_user_id=user_id,
|
|
)
|
|
db.add(rule)
|
|
else:
|
|
rule.category_id = category_id
|
|
db.flush()
|
|
return rule
|