T2: Pydantic v2 — replace @dataclass with BaseModel
- All domain models now use pydantic.BaseModel
- Removed manual to_dict()/from_dict() — use model_dump(mode='json')/model_validate()
- Added field validators:
- Account.balance >= 0
- Transaction.amount != 0
- RecurringCashflow.amount != 0, frequency in {daily, weekly, monthly, yearly}
- Liability.interest, payment >= 0
- ExchangeRate.rate > 0
- ForecastScenario.multipliers >= 0
- pydantic v2 native UUID handling (auto str in JSON)
- Updated excel_sync.py to use model_validate()
- Updated tests/test_model.py and tests/test_currency.py
- Added pydantic>=2.0 to pyproject.toml
Tests: 63/63 pass.
This commit is contained in:
+34
-29
@@ -1,7 +1,8 @@
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from domain.account import Account
|
||||
from domain.asset import Asset
|
||||
from domain.currency import DEFAULT_RATES, ExchangeRate
|
||||
@@ -11,37 +12,41 @@ from domain.scenario import ForecastScenario
|
||||
from domain.transaction import Transaction
|
||||
|
||||
|
||||
@dataclass
|
||||
class FinancialModel:
|
||||
base_currency: str = "RUB"
|
||||
accounts: list[Account] = field(default_factory=list)
|
||||
transactions: list[Transaction] = field(default_factory=list)
|
||||
recurring: list[RecurringCashflow] = field(default_factory=list)
|
||||
assets: list[Asset] = field(default_factory=list)
|
||||
liabilities: list[Liability] = field(default_factory=list)
|
||||
scenarios: list[ForecastScenario] = field(default_factory=list)
|
||||
exchange_rates: list[ExchangeRate] = field(default_factory=lambda: DEFAULT_RATES.copy())
|
||||
class FinancialModel(BaseModel):
|
||||
"""Корневая модель финансового плана."""
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
SCHEMA_VERSION: int = 1 # NB: не Field — это class-level metadata, не pydantic field
|
||||
|
||||
base_currency: str = "RUB"
|
||||
accounts: list[Account] = Field(default_factory=list)
|
||||
transactions: list[Transaction] = Field(default_factory=list)
|
||||
recurring: list[RecurringCashflow] = Field(default_factory=list)
|
||||
assets: list[Asset] = Field(default_factory=list)
|
||||
liabilities: list[Liability] = Field(default_factory=list)
|
||||
scenarios: list[ForecastScenario] = Field(default_factory=list)
|
||||
exchange_rates: list[ExchangeRate] = Field(
|
||||
default_factory=lambda: [r.model_copy() for r in DEFAULT_RATES]
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Сериализация в dict с version и UUID-as-str (для JSON)."""
|
||||
return {
|
||||
"version": self.SCHEMA_VERSION,
|
||||
"base_currency": self.base_currency,
|
||||
"accounts": [a.to_dict() for a in self.accounts],
|
||||
"transactions": [t.to_dict() for t in self.transactions],
|
||||
"recurring": [r.to_dict() for r in self.recurring],
|
||||
"assets": [a.to_dict() for a in self.assets],
|
||||
"liabilities": [li.to_dict() for li in self.liabilities],
|
||||
"scenarios": [s.to_dict() for s in self.scenarios],
|
||||
"exchange_rates": [r.to_dict() for r in self.exchange_rates],
|
||||
"accounts": [a.model_dump(mode="json") for a in self.accounts],
|
||||
"transactions": [t.model_dump(mode="json") for t in self.transactions],
|
||||
"recurring": [r.model_dump(mode="json") for r in self.recurring],
|
||||
"assets": [a.model_dump(mode="json") for a in self.assets],
|
||||
"liabilities": [li.model_dump(mode="json") for li in self.liabilities],
|
||||
"scenarios": [s.model_dump(mode="json") for s in self.scenarios],
|
||||
"exchange_rates": [e.model_dump(mode="json") for e in self.exchange_rates],
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "FinancialModel":
|
||||
version = data.get("version", 0)
|
||||
if version == 0:
|
||||
return cls._from_v0(data)
|
||||
return cls._from_legacy(data)
|
||||
if version == 1:
|
||||
return cls._from_v1(data)
|
||||
raise ValueError(f"Unsupported FinancialModel version: {version}")
|
||||
@@ -50,18 +55,18 @@ class FinancialModel:
|
||||
def _from_v1(cls, data: dict) -> "FinancialModel":
|
||||
return cls(
|
||||
base_currency=data.get("base_currency", "RUB"),
|
||||
accounts=[Account.from_dict(a) for a in data.get("accounts", [])],
|
||||
transactions=[Transaction.from_dict(t) for t in data.get("transactions", [])],
|
||||
recurring=[RecurringCashflow.from_dict(r) for r in data.get("recurring", [])],
|
||||
assets=[Asset.from_dict(a) for a in data.get("assets", [])],
|
||||
liabilities=[Liability.from_dict(li) for li in data.get("liabilities", [])],
|
||||
scenarios=[ForecastScenario.from_dict(s) for s in data.get("scenarios", [])],
|
||||
exchange_rates=[ExchangeRate.from_dict(r) for r in data.get("exchange_rates", [])],
|
||||
accounts=[Account.model_validate(a) for a in data.get("accounts", [])],
|
||||
transactions=[Transaction.model_validate(t) for t in data.get("transactions", [])],
|
||||
recurring=[RecurringCashflow.model_validate(r) for r in data.get("recurring", [])],
|
||||
assets=[Asset.model_validate(a) for a in data.get("assets", [])],
|
||||
liabilities=[Liability.model_validate(li) for li in data.get("liabilities", [])],
|
||||
scenarios=[ForecastScenario.model_validate(s) for s in data.get("scenarios", [])],
|
||||
exchange_rates=[ExchangeRate.model_validate(r) for r in data.get("exchange_rates", [])],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _from_v0(cls, data: dict) -> "FinancialModel":
|
||||
# Legacy: файлы, сохранённые до введения version. Совпадает по структуре с v1.
|
||||
def _from_legacy(cls, data: dict) -> "FinancialModel":
|
||||
# Legacy: файлы без version. Структура совпадает с v1.
|
||||
return cls._from_v1(data)
|
||||
|
||||
def save(self, path: str | Path) -> None:
|
||||
|
||||
Reference in New Issue
Block a user