Storage

SQLite-backed storage layer for receipts and counterparties.

finamt.storage.base

finamt.storage.base

Abstract repository interface.

class finamt.storage.base.ReceiptRepository(*args, **kwargs)

Bases: Protocol

Storage abstraction for receipt persistence.

save(receipt: ReceiptData) → bool

Persist a receipt across all tables.

Returns True if saved, False if a duplicate already exists (same content hash). Callers can distinguish via the return value rather than catching an exception.

get(receipt_id: str) → ReceiptData | None

Fetch a receipt by content-hash ID.

exists(receipt_id: str) → bool

Return True if a receipt with this ID is already stored.

delete(receipt_id: str) → bool

Remove a receipt and all its child rows. Returns True if deleted.

list_all() → Iterable[ReceiptData]

All receipts, most recently dated first.

find_by_period(start: date, end: date) → Iterable[ReceiptData]

Receipts whose date falls within [start, end] inclusive.

find_by_category(category: str) → Iterable[ReceiptData]

Receipts matching the given category.

find_by_type(receipt_type: str) → Iterable[ReceiptData]

Receipts of a given type.

Parameters:

receipt_type – "purchase" or "sale".

get_or_create_counterparty(counterparty: Counterparty) → Counterparty

Return the existing counterparty if one matches by VAT ID or name, otherwise insert and return a new one.

relink_counterparty(receipt_id: str, fields: dict) → bool

Find-or-create a counterparty by name/VAT-ID and link only this receipt to it. The previous counterparty row is untouched. Returns True if the receipt was found.

close() → None

Release connections.

finamt.storage.sqlite

finamt.storage.sqlite

SQLite-backed receipt repository — 4-table schema.

Tables

counterparties — vendors and clients with parsed address + tax numbers receipts — core record: hash id, FK to counterparty, type, totals receipt_items — line items, FK to receipt receipt_content — raw OCR text, FK to receipt (kept separate, can be large)

Receipt ID

The id is the SHA-256 hash of the normalised OCR text (computed by ReceiptData.__post_init__). Identical content → identical ID → duplicate.

Default path: ~/.finamt/default/finamt.db

class finamt.storage.sqlite.SQLiteRepository(db_path: Path | str | None = None)

Bases: object

Persistent SQLite storage implementing ReceiptRepository.

close() → None

get_or_create_counterparty(cp: Counterparty) → Counterparty

Return an existing counterparty matching by name only (case-insensitive).

VAT-ID is intentionally NOT used as a match key: agent OCR errors can produce the same VAT ID for completely different companies (e.g. the taxpayer’s own ID being attached to a supplier), and merging on VAT ID alone would silently overwrite unrelated counterparties. Duplicate VAT IDs are surfaced to the user in the UI instead.

Only inserts a new row when no name-match is found. The SELECT + INSERT is performed under the write lock to prevent duplicate rows from concurrent uploads.

save(receipt: ReceiptData) → bool

Persist a receipt.

Returns True on success, False if a duplicate already exists. Raises no exceptions on duplicate — callers check the return value or call exists() first.

get_postings(receipt_id: str) → list[Posting]

Return all postings for receipt_id, ordered by position.

list_all_postings() → list[dict]

Return all postings across all receipts as dicts (e.g. for EÜR derivation).

exists(receipt_id: str) → bool

get(receipt_id: str) → ReceiptData | None

delete(receipt_id: str) → bool

update(receipt_id: str, fields: dict) → bool

Partially update a receipt’s mutable fields (user corrections).

Receipt fields: receipt_type, receipt_number, receipt_date, total_amount, vat_percentage, vat_amount, category.

Counterparty fields (applied to the counterparty row owned by this receipt): counterparty_name, vat_id, tax_number, and address sub-fields via an address dict with keys street_and_number, address_supplement, postcode, city, state, country.

Returns True if the receipt row was found.

list_verified_counterparties() → list[dict]

Return all verified counterparties sorted alphabetically by name (case-insensitive).

get_category_defaults_for_counterparty(cp_id: str) → dict

Return the most-used (category, subcategory) pair for a counterparty.

Prefers non-‘other’ categories. Returns empty dict when no receipts exist yet for this counterparty.

set_counterparty_verified(cp_id: str, verified: bool) → None

list_all_counterparties() → list[dict]

Return every counterparty row sorted alphabetically by name (case-insensitive).

update_counterparty(cp_id: str, fields: dict) → bool

Update editable fields of a counterparty. Returns True if a row was updated.

The verified flag is only changed when explicitly included in fields; ordinary field edits (name, address, VAT-ID, …) must not touch it.

relink_counterparty(receipt_id: str, fields: dict) → bool

Find-or-create a counterparty by name/VAT-ID and link only this receipt to it.

The old counterparty row is untouched — if it becomes unreferenced the startup orphan-cleanup will remove it on the next open. Returns True if the receipt row was found and updated.

delete_counterparty(cp_id: str) → bool

Delete a counterparty by id. Returns True if a row was removed.

list_all() → Iterable[ReceiptData]

find_by_period(start: date, end: date) → Iterable[ReceiptData]

find_by_category(category: str) → Iterable[ReceiptData]

find_by_type(receipt_type: str) → Iterable[ReceiptData]

get_metadata(key: str) → dict | None

Return the parsed JSON value for key, or None if not set.

set_metadata(key: str, value: dict) → None

Upsert key with the JSON-serialised value.

delete_metadata(key: str) → None

Remove key from project_metadata (no-op if absent).

finamt.storage.project

finamt.storage.project

Project layout resolution — maps a project name to its three paths:

~/.finamt/<project>/finamt.db — SQLite database ~/.finamt/<project>/pdfs/ — original PDF archive ~/.finamt/<project>/debug/ — per-receipt agent debug output

Usage:

from finamt.storage.project import resolve_project, layout_from_db_path

layout = resolve_project()                   # uses "default" or FINAMT_PROJECT env var
layout = resolve_project("acme-gmbh-2025")   # explicit project name
layout = layout_from_db_path(Path("..."))    # reverse: infer layout from db path

Migration note (breaking change in 0.x)

Previous versions stored data at ~/.finamt/finamt.db (flat layout). The new structure requires data to live under a project subfolder. To migrate an existing installation:

mkdir -p ~/.finamt/default
mv ~/.finamt/finamt.db ~/.finamt/default/
mv ~/.finamt/pdfs         ~/.finamt/default/
mv ~/.finamt/debug        ~/.finamt/default/

class finamt.storage.project.ProjectLayout(name: str, root: Path, db_path: Path, pdfs_dir: Path, debug_dir: Path)

Bases: object

All paths belonging to a single project.

name: str

root: Path

db_path: Path

pdfs_dir: Path

debug_dir: Path

create_dirs() → None

Ensure all project directories exist.

property is_default: bool

property exists: bool

True if the db file has been created.

finamt.storage.project.resolve_project(project: str | None = None, *, env_var: bool = True) → ProjectLayout

Resolve a project name to its layout.

Priority order:

1. Explicit project argument

2. FINAMT_PROJECT environment variable (when env_var=True)

3. "default"

finamt.storage.project.layout_from_db_path(db_path: Path) → ProjectLayout

Infer a ProjectLayout from an explicit db path.

If the db lives at ~/.finamt/<name>/finamt.db the project name is taken from the containing directory. Any other path uses the db file’s stem as the project name, rooting everything in the db’s parent directory.

finamt.storage.project.validate_project_name(name: str) → str | None

Validate a proposed project name. Returns an error message string on failure, None on success.

finamt.storage.project.list_projects() → list[ProjectLayout]

Scan FINAMT_HOME for project subdirectories. Returns layouts sorted: default first, then alphabetically.