feat: add SQLAlchemy database models and tests
Add the data layer for the spending analysis app including models for household members, accounts, categories, transactions, categorization rules, and CSV import mappings. All models use SQLAlchemy 2.0 mapped columns with proper foreign key relationships. Includes db.py with Base class, engine/session factories, and 6 passing tests. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
21
src/db.py
Normal file
21
src/db.py
Normal file
@@ -0,0 +1,21 @@
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import DeclarativeBase, Session
|
||||
|
||||
|
||||
class Base(DeclarativeBase):
|
||||
pass
|
||||
|
||||
|
||||
def get_engine(db_path: Path | None = None):
|
||||
if db_path is None:
|
||||
db_path = Path.home() / ".spending_analysis" / "spending.db"
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
return create_engine(f"sqlite:///{db_path}")
|
||||
|
||||
|
||||
def get_session(db_path: Path | None = None) -> Session:
|
||||
engine = get_engine(db_path)
|
||||
Base.metadata.create_all(engine)
|
||||
return Session(engine)
|
||||
@@ -0,0 +1,6 @@
|
||||
from src.models.household import HouseholdMember
|
||||
from src.models.account import Account
|
||||
from src.models.category import Category
|
||||
from src.models.transaction import Transaction
|
||||
from src.models.rule import CategorizationRule
|
||||
from src.models.csv_mapping import CsvMapping
|
||||
|
||||
24
src/models/account.py
Normal file
24
src/models/account.py
Normal file
@@ -0,0 +1,24 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import ForeignKey, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from src.db import Base
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from src.models.household import HouseholdMember
|
||||
|
||||
|
||||
class Account(Base):
|
||||
__tablename__ = "accounts"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
name: Mapped[str] = mapped_column(String(100))
|
||||
institution: Mapped[str] = mapped_column(String(100))
|
||||
account_type: Mapped[str] = mapped_column(String(20))
|
||||
owner_id: Mapped[int | None] = mapped_column(ForeignKey("household_members.id"))
|
||||
is_shared: Mapped[bool] = mapped_column(default=False)
|
||||
|
||||
owner: Mapped[HouseholdMember | None] = relationship(back_populates="accounts")
|
||||
13
src/models/category.py
Normal file
13
src/models/category.py
Normal file
@@ -0,0 +1,13 @@
|
||||
from sqlalchemy import String
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from src.db import Base
|
||||
|
||||
|
||||
class Category(Base):
|
||||
__tablename__ = "categories"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
name: Mapped[str] = mapped_column(String(100), unique=True)
|
||||
default_tag: Mapped[str | None] = mapped_column(String(20))
|
||||
icon: Mapped[str | None] = mapped_column(String(50))
|
||||
24
src/models/csv_mapping.py
Normal file
24
src/models/csv_mapping.py
Normal file
@@ -0,0 +1,24 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import ForeignKey, String, Text
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from src.db import Base
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from src.models.account import Account
|
||||
|
||||
|
||||
class CsvMapping(Base):
|
||||
__tablename__ = "csv_mappings"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
name: Mapped[str] = mapped_column(String(100))
|
||||
fingerprint: Mapped[str] = mapped_column(String(500))
|
||||
column_map: Mapped[str] = mapped_column(Text)
|
||||
amount_logic: Mapped[str] = mapped_column(String(50))
|
||||
account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"))
|
||||
|
||||
account: Mapped[Account] = relationship()
|
||||
21
src/models/household.py
Normal file
21
src/models/household.py
Normal file
@@ -0,0 +1,21 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import String
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship as sa_relationship
|
||||
|
||||
from src.db import Base
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from src.models.account import Account
|
||||
|
||||
|
||||
class HouseholdMember(Base):
|
||||
__tablename__ = "household_members"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
name: Mapped[str] = mapped_column(String(100))
|
||||
relationship: Mapped[str] = mapped_column(String(50))
|
||||
|
||||
accounts: Mapped[list[Account]] = sa_relationship(back_populates="owner")
|
||||
26
src/models/rule.py
Normal file
26
src/models/rule.py
Normal file
@@ -0,0 +1,26 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import ForeignKey, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from src.db import Base
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from src.models.category import Category
|
||||
from src.models.household import HouseholdMember
|
||||
|
||||
|
||||
class CategorizationRule(Base):
|
||||
__tablename__ = "categorization_rules"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
pattern: Mapped[str] = mapped_column(String(300))
|
||||
category_id: Mapped[int] = mapped_column(ForeignKey("categories.id"))
|
||||
tag_override: Mapped[str | None] = mapped_column(String(20))
|
||||
attributed_to_id: Mapped[int | None] = mapped_column(ForeignKey("household_members.id"))
|
||||
priority: Mapped[int] = mapped_column(default=0)
|
||||
|
||||
category: Mapped[Category] = relationship()
|
||||
attributed_to: Mapped[HouseholdMember | None] = relationship()
|
||||
35
src/models/transaction.py
Normal file
35
src/models/transaction.py
Normal file
@@ -0,0 +1,35 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import Date, ForeignKey, Numeric, String, Text
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from src.db import Base
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from src.models.account import Account
|
||||
from src.models.category import Category
|
||||
from src.models.household import HouseholdMember
|
||||
|
||||
|
||||
class Transaction(Base):
|
||||
__tablename__ = "transactions"
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
date: Mapped[datetime.date] = mapped_column(Date)
|
||||
amount: Mapped[float] = mapped_column(Numeric(10, 2))
|
||||
description: Mapped[str] = mapped_column(String(300))
|
||||
raw_description: Mapped[str | None] = mapped_column(Text)
|
||||
account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"))
|
||||
category_id: Mapped[int | None] = mapped_column(ForeignKey("categories.id"))
|
||||
attributed_to_id: Mapped[int | None] = mapped_column(ForeignKey("household_members.id"))
|
||||
tag: Mapped[str | None] = mapped_column(String(20))
|
||||
source_category: Mapped[str | None] = mapped_column(String(100))
|
||||
is_transfer: Mapped[bool] = mapped_column(default=False)
|
||||
transfer_pair_id: Mapped[int | None] = mapped_column(ForeignKey("transactions.id"))
|
||||
|
||||
account: Mapped[Account] = relationship()
|
||||
category: Mapped[Category | None] = relationship()
|
||||
attributed_to: Mapped[HouseholdMember | None] = relationship()
|
||||
112
tests/models/test_models.py
Normal file
112
tests/models/test_models.py
Normal file
@@ -0,0 +1,112 @@
|
||||
import datetime
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.db import Base
|
||||
from src.models.household import HouseholdMember
|
||||
from src.models.account import Account
|
||||
from src.models.category import Category
|
||||
from src.models.transaction import Transaction
|
||||
from src.models.rule import CategorizationRule
|
||||
from src.models.csv_mapping import CsvMapping
|
||||
|
||||
|
||||
def make_session():
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(engine)
|
||||
return Session(engine)
|
||||
|
||||
|
||||
def test_create_household_member():
|
||||
session = make_session()
|
||||
member = HouseholdMember(name="Andrew", relationship="self")
|
||||
session.add(member)
|
||||
session.commit()
|
||||
assert member.id is not None
|
||||
assert member.name == "Andrew"
|
||||
|
||||
|
||||
def test_create_account_with_owner():
|
||||
session = make_session()
|
||||
member = HouseholdMember(name="Andrew", relationship="self")
|
||||
session.add(member)
|
||||
session.flush()
|
||||
account = Account(
|
||||
name="Chase Freedom",
|
||||
institution="Chase",
|
||||
account_type="credit",
|
||||
owner_id=member.id,
|
||||
)
|
||||
session.add(account)
|
||||
session.commit()
|
||||
assert account.owner.name == "Andrew"
|
||||
|
||||
|
||||
def test_create_category():
|
||||
session = make_session()
|
||||
cat = Category(name="Groceries", default_tag="needs")
|
||||
session.add(cat)
|
||||
session.commit()
|
||||
assert cat.id is not None
|
||||
|
||||
|
||||
def test_create_transaction():
|
||||
session = make_session()
|
||||
member = HouseholdMember(name="Andrew", relationship="self")
|
||||
session.add(member)
|
||||
session.flush()
|
||||
account = Account(name="Checking", institution="Wells Fargo", account_type="checking", owner_id=member.id)
|
||||
cat = Category(name="Groceries", default_tag="needs")
|
||||
session.add_all([account, cat])
|
||||
session.flush()
|
||||
txn = Transaction(
|
||||
date=datetime.date(2026, 1, 15),
|
||||
amount=-48.52,
|
||||
description="WAL-MART #7181",
|
||||
raw_description="PURCHASE AUTHORIZED ON 01/14 WAL-MART #7181 BEAUFORT SC CARD 5360",
|
||||
account_id=account.id,
|
||||
category_id=cat.id,
|
||||
attributed_to_id=member.id,
|
||||
tag="needs",
|
||||
)
|
||||
session.add(txn)
|
||||
session.commit()
|
||||
assert txn.id is not None
|
||||
assert txn.account.name == "Checking"
|
||||
assert txn.category.name == "Groceries"
|
||||
|
||||
|
||||
def test_create_categorization_rule():
|
||||
session = make_session()
|
||||
cat = Category(name="Groceries", default_tag="needs")
|
||||
session.add(cat)
|
||||
session.flush()
|
||||
rule = CategorizationRule(
|
||||
pattern="WAL-MART",
|
||||
category_id=cat.id,
|
||||
priority=10,
|
||||
)
|
||||
session.add(rule)
|
||||
session.commit()
|
||||
assert rule.id is not None
|
||||
|
||||
|
||||
def test_create_csv_mapping():
|
||||
session = make_session()
|
||||
member = HouseholdMember(name="Andrew", relationship="self")
|
||||
session.add(member)
|
||||
session.flush()
|
||||
account = Account(name="Chase", institution="Chase", account_type="credit", owner_id=member.id)
|
||||
session.add(account)
|
||||
session.flush()
|
||||
mapping = CsvMapping(
|
||||
name="Chase Credit Card",
|
||||
fingerprint="Transaction Date,Post Date,Description,Category,Type,Amount,Memo",
|
||||
column_map='{"date": "Transaction Date", "amount": "Amount", "description": "Description"}',
|
||||
amount_logic="signed",
|
||||
account_id=account.id,
|
||||
)
|
||||
session.add(mapping)
|
||||
session.commit()
|
||||
assert mapping.id is not None
|
||||
Reference in New Issue
Block a user