Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 20 additions & 3 deletions src/lampyrid/models/firefly_models.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,28 @@
# generated by datamodel-codegen:
# filename: firefly-iii-6.5.5-v1.yaml
# timestamp: 2026-03-24T13:57:48+00:00
#
# NOTE: Manual modifications have been made to this file to work around
# Firefly III API bugs. See comments marked with "MANUAL FIX" below.
# These fixes are re-applied automatically after regeneration by
# scripts/update_schema.py (apply_manual_fixes).

from __future__ import annotations

from datetime import date as date_aliased
from enum import Enum
from typing import Any
from typing import Annotated, Any

from pydantic import AnyUrl, AwareDatetime, BaseModel, EmailStr, Field, RootModel
from pydantic import AnyUrl, AwareDatetime, BaseModel, BeforeValidator, EmailStr, Field, RootModel


# MANUAL FIX: Firefly III API bug (Issue #43) returns currency_id as int instead of str
# in spent/pc_spent and other ArrayEntryWithCurrencyAndSum arrays. This validator coerces
# int to str so model validation does not fail.
def _coerce_to_str(v: Any) -> str | None:
if v is None:
return None
return str(v)


class AutocompleteAccount(BaseModel):
Expand Down Expand Up @@ -521,7 +535,10 @@ class InsightTransferEntry(BaseModel):


class ArrayEntryWithCurrencyAndSum(BaseModel):
currency_id: str | None = Field(None, examples=['5'])
# MANUAL FIX: Use Annotated with BeforeValidator to coerce int to str (Issue #43)
currency_id: Annotated[str | None, BeforeValidator(_coerce_to_str)] = Field(
None, examples=['5']
)
currency_code: str | None = Field(None, examples=['USD'])
currency_symbol: str | None = Field(None, examples=['$'])
currency_decimal_places: int | None = Field(
Expand Down
99 changes: 98 additions & 1 deletion src/lampyrid/scripts/update_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -159,8 +159,100 @@ def update_pyproject_toml(old_version: str | None, new_version: str) -> bool:
return False


# Manual patches re-applied to the generated models file after every regeneration.
# Each entry is (anchor, replacement): ``anchor`` is the exact generated snippet that
# must be present, ``replacement`` is what it becomes. If an anchor is missing,
# datamodel-codegen's output has drifted and apply_manual_fixes() fails loudly rather
# than silently dropping a workaround for a known Firefly III API bug.
_MANUAL_FIXES: list[tuple[str, str]] = [
# Issue #43: import the symbols the coercion validator needs and define it.
(
'from typing import Any\n',
'from typing import Annotated, Any\n',
),
(
'from pydantic import AnyUrl, AwareDatetime, BaseModel, EmailStr, Field, RootModel\n',
'from pydantic import (\n'
' AnyUrl,\n'
' AwareDatetime,\n'
' BaseModel,\n'
' BeforeValidator,\n'
' EmailStr,\n'
' Field,\n'
' RootModel,\n'
')\n'
'\n\n'
'# MANUAL FIX: Firefly III API bug (Issue #43) returns currency_id as int instead of\n'
'# str in spent/pc_spent (ArrayEntryWithCurrencyAndSum) arrays. Coerce int to str.\n'
'def _coerce_to_str(v):\n'
' if v is None:\n'
' return None\n'
' return str(v)\n',
),
# Issue #43: wrap ArrayEntryWithCurrencyAndSum.currency_id with the coercing validator.
(
'class ArrayEntryWithCurrencyAndSum(BaseModel):\n'
" currency_id: str | None = Field(None, examples=['5'])\n",
'class ArrayEntryWithCurrencyAndSum(BaseModel):\n'
' # MANUAL FIX: coerce int to str (Issue #43)\n'
' currency_id: Annotated[str | None, BeforeValidator(_coerce_to_str)] = Field(\n'
" None, examples=['5']\n"
' )\n',
),
]

# Marker proving the patches are present, used as the idempotency guard.
_MANUAL_FIX_MARKER = '_coerce_to_str'


def apply_manual_fixes() -> bool:
"""Re-apply manual patches to the freshly generated models file.

datamodel-codegen overwrites firefly_models.py from the OpenAPI spec, which
drops any hand-written workarounds for known Firefly III API bugs. This
re-injects them idempotently so they survive every regeneration.

Currently applied fixes:
- Issue #43: the API returns ``currency_id`` as an int in spent/pc_spent
(ArrayEntryWithCurrencyAndSum) arrays, but the spec types it as a string.
A ``BeforeValidator`` coerces int -> str so validation does not fail.

Returns:
True if the file was patched, False if fixes were already present.

Raises:
FileNotFoundError: if the generated models file does not exist.
RuntimeError: if a patch anchor is missing, meaning datamodel-codegen's
output drifted and the workaround would otherwise be silently lost.

"""
models_path = PROJECT_ROOT / MODELS_OUTPUT
if not models_path.exists():
raise FileNotFoundError(f'{MODELS_OUTPUT} not found, cannot apply manual fixes')

content = models_path.read_text()

# Idempotency guard: if the marker is already present, the file is patched.
if _MANUAL_FIX_MARKER in content:
return False

for anchor, replacement in _MANUAL_FIXES:
if anchor not in content:
raise RuntimeError(
f'Manual fix anchor not found in {MODELS_OUTPUT}; datamodel-codegen '
f'output likely changed. Missing anchor:\n{anchor!r}'
)
content = content.replace(anchor, replacement, 1)

models_path.write_text(content)
return True


def regenerate_models() -> bool:
"""Run datamodel-codegen to regenerate Pydantic models.
"""Run datamodel-codegen to regenerate Pydantic models, then re-apply fixes.

Manual workarounds for known Firefly III API bugs are re-applied here (not only
on the CLI path) so every caller of this helper gets a patched models file.

Returns:
True if successful, False otherwise
Expand All @@ -174,10 +266,15 @@ def regenerate_models() -> bool:
text=True,
check=True,
)
if apply_manual_fixes():
print(' Applied manual fixes for known Firefly III API bugs')
return True
except subprocess.CalledProcessError as e:
print(f' Error running datamodel-codegen: {e.stderr}')
return False
except (FileNotFoundError, RuntimeError) as e:
print(f' Error applying manual fixes: {e}')
return False


def cleanup_old_schema(old_version: str | None, new_version: str) -> bool:
Expand Down
78 changes: 78 additions & 0 deletions tests/unit/test_scripts.py
Original file line number Diff line number Diff line change
Expand Up @@ -260,3 +260,81 @@ def test_regenerate_models_failure(self, mock_run):
result = update_schema.regenerate_models()

assert result is False

# --- apply_manual_fixes -------------------------------------------------

# Minimal stand-in for freshly generated firefly_models.py containing the
# exact anchors apply_manual_fixes() patches.
_GENERATED_STUB = (
'from __future__ import annotations\n\n'
'from typing import Any\n\n'
'from pydantic import AnyUrl, AwareDatetime, BaseModel, EmailStr, Field, RootModel\n\n\n'
'class ArrayEntryWithCurrencyAndSum(BaseModel):\n'
" currency_id: str | None = Field(None, examples=['5'])\n"
)

def _stub_models_file(self, tmp_path, monkeypatch, content):
"""Point update_schema at a temp models file containing ``content``."""
models_path = tmp_path / 'firefly_models.py'
models_path.write_text(content)
monkeypatch.setattr(update_schema, 'PROJECT_ROOT', tmp_path)
monkeypatch.setattr(update_schema, 'MODELS_OUTPUT', 'firefly_models.py')
return models_path

def test_apply_manual_fixes_patches_generated_file(self, tmp_path, monkeypatch):
"""Fresh generated content is patched with the int->str coercion."""
models_path = self._stub_models_file(tmp_path, monkeypatch, self._GENERATED_STUB)

assert update_schema.apply_manual_fixes() is True

patched = models_path.read_text()
assert '_coerce_to_str' in patched
assert 'BeforeValidator' in patched
assert 'Annotated[str | None, BeforeValidator(_coerce_to_str)]' in patched

def test_apply_manual_fixes_is_idempotent(self, tmp_path, monkeypatch):
"""Already-patched content is left untouched and reported as no-op."""
models_path = self._stub_models_file(tmp_path, monkeypatch, self._GENERATED_STUB)
update_schema.apply_manual_fixes()
patched = models_path.read_text()

assert update_schema.apply_manual_fixes() is False
assert models_path.read_text() == patched

def test_apply_manual_fixes_missing_anchor_raises(self, tmp_path, monkeypatch):
"""A drifted generated file (anchor missing) fails loudly."""
drifted = self._GENERATED_STUB.replace(
'from typing import Any\n', 'from typing import List\n'
)
self._stub_models_file(tmp_path, monkeypatch, drifted)

with pytest.raises(RuntimeError, match='anchor not found'):
update_schema.apply_manual_fixes()

def test_apply_manual_fixes_missing_file_raises(self, tmp_path, monkeypatch):
"""Missing models file fails loudly rather than silently skipping."""
monkeypatch.setattr(update_schema, 'PROJECT_ROOT', tmp_path)
monkeypatch.setattr(update_schema, 'MODELS_OUTPUT', 'does_not_exist.py')

with pytest.raises(FileNotFoundError):
update_schema.apply_manual_fixes()

@patch('subprocess.run')
def test_regenerate_models_applies_fixes(self, mock_run, tmp_path, monkeypatch):
"""regenerate_models patches the file after a successful codegen run."""
mock_run.return_value = MagicMock(returncode=0)
models_path = self._stub_models_file(tmp_path, monkeypatch, self._GENERATED_STUB)

assert update_schema.regenerate_models() is True
assert '_coerce_to_str' in models_path.read_text()

@patch('subprocess.run')
def test_regenerate_models_fails_loud_on_drift(self, mock_run, tmp_path, monkeypatch):
"""regenerate_models returns False if manual fixes cannot be applied."""
mock_run.return_value = MagicMock(returncode=0)
drifted = self._GENERATED_STUB.replace(
'from typing import Any\n', 'from typing import List\n'
)
self._stub_models_file(tmp_path, monkeypatch, drifted)

assert update_schema.regenerate_models() is False
Loading