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:
+9
-19
@@ -1,27 +1,17 @@
|
||||
from dataclasses import dataclass, field
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
@dataclass
|
||||
class Account:
|
||||
id: UUID = field(default_factory=uuid4)
|
||||
|
||||
class Account(BaseModel):
|
||||
id: UUID = Field(default_factory=uuid4)
|
||||
name: str = ""
|
||||
currency: str = "USD"
|
||||
balance: float = 0.0
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": str(self.id),
|
||||
"name": self.name,
|
||||
"currency": self.currency,
|
||||
"balance": self.balance,
|
||||
}
|
||||
|
||||
@field_validator("balance")
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Account":
|
||||
return cls(
|
||||
id=UUID(data["id"]),
|
||||
name=data["name"],
|
||||
currency=data.get("currency", "USD"),
|
||||
balance=data.get("balance", 0.0),
|
||||
)
|
||||
def _balance_non_negative(cls, v: float) -> float:
|
||||
if v < 0:
|
||||
raise ValueError("balance must be non-negative")
|
||||
return v
|
||||
|
||||
+4
-21
@@ -1,27 +1,10 @@
|
||||
from dataclasses import dataclass, field
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
@dataclass
|
||||
class Asset:
|
||||
id: UUID = field(default_factory=uuid4)
|
||||
|
||||
class Asset(BaseModel):
|
||||
id: UUID = Field(default_factory=uuid4)
|
||||
name: str = ""
|
||||
value: float = 0.0
|
||||
growth_rate: float = 0.0
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": str(self.id),
|
||||
"name": self.name,
|
||||
"value": self.value,
|
||||
"growth_rate": self.growth_rate,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Asset":
|
||||
return cls(
|
||||
id=UUID(data["id"]),
|
||||
name=data["name"],
|
||||
value=data.get("value", 0.0),
|
||||
growth_rate=data.get("growth_rate", 0.0),
|
||||
)
|
||||
|
||||
+7
-16
@@ -1,4 +1,4 @@
|
||||
from dataclasses import dataclass
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
CURRENCY_SYMBOLS = {
|
||||
"RUB": "₽",
|
||||
@@ -12,26 +12,17 @@ CURRENCY_SYMBOLS = {
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExchangeRate:
|
||||
class ExchangeRate(BaseModel):
|
||||
from_currency: str = "USD"
|
||||
to_currency: str = "RUB"
|
||||
rate: float = 80.0
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"from_currency": self.from_currency,
|
||||
"to_currency": self.to_currency,
|
||||
"rate": self.rate,
|
||||
}
|
||||
|
||||
@field_validator("rate")
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ExchangeRate":
|
||||
return cls(
|
||||
from_currency=data.get("from_currency", "USD"),
|
||||
to_currency=data.get("to_currency", "RUB"),
|
||||
rate=data.get("rate", 80.0),
|
||||
)
|
||||
def _rate_positive(cls, v: float) -> float:
|
||||
if v <= 0:
|
||||
raise ValueError("rate must be positive")
|
||||
return v
|
||||
|
||||
|
||||
DEFAULT_RATES: list[ExchangeRate] = [
|
||||
|
||||
+9
-21
@@ -1,30 +1,18 @@
|
||||
from dataclasses import dataclass, field
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
@dataclass
|
||||
class Liability:
|
||||
id: UUID = field(default_factory=uuid4)
|
||||
|
||||
class Liability(BaseModel):
|
||||
id: UUID = Field(default_factory=uuid4)
|
||||
name: str = ""
|
||||
balance: float = 0.0
|
||||
interest: float = 0.0
|
||||
payment: float = 0.0
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": str(self.id),
|
||||
"name": self.name,
|
||||
"balance": self.balance,
|
||||
"interest": self.interest,
|
||||
"payment": self.payment,
|
||||
}
|
||||
|
||||
@field_validator("interest", "payment")
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Liability":
|
||||
return cls(
|
||||
id=UUID(data["id"]),
|
||||
name=data["name"],
|
||||
balance=data.get("balance", 0.0),
|
||||
interest=data.get("interest", 0.0),
|
||||
payment=data.get("payment", 0.0),
|
||||
)
|
||||
def _non_negative(cls, v: float) -> float:
|
||||
if v < 0:
|
||||
raise ValueError("must be non-negative")
|
||||
return v
|
||||
|
||||
+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:
|
||||
|
||||
+16
-23
@@ -1,33 +1,26 @@
|
||||
from dataclasses import dataclass, field
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
@dataclass
|
||||
class RecurringCashflow:
|
||||
id: UUID = field(default_factory=uuid4)
|
||||
|
||||
class RecurringCashflow(BaseModel):
|
||||
id: UUID = Field(default_factory=uuid4)
|
||||
start_date: str = ""
|
||||
end_date: str = ""
|
||||
frequency: str = "monthly"
|
||||
amount: float = 0.0
|
||||
category: str = ""
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": str(self.id),
|
||||
"start_date": self.start_date,
|
||||
"end_date": self.end_date,
|
||||
"frequency": self.frequency,
|
||||
"amount": self.amount,
|
||||
"category": self.category,
|
||||
}
|
||||
|
||||
@field_validator("frequency")
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "RecurringCashflow":
|
||||
return cls(
|
||||
id=UUID(data["id"]),
|
||||
start_date=data.get("start_date", ""),
|
||||
end_date=data.get("end_date", ""),
|
||||
frequency=data.get("frequency", "monthly"),
|
||||
amount=data.get("amount", 0.0),
|
||||
category=data.get("category", ""),
|
||||
)
|
||||
def _frequency_known(cls, v: str) -> str:
|
||||
if v not in {"daily", "weekly", "monthly", "yearly"}:
|
||||
raise ValueError(f"unknown frequency: {v}")
|
||||
return v
|
||||
|
||||
@field_validator("amount")
|
||||
@classmethod
|
||||
def _amount_nonzero(cls, v: float) -> float:
|
||||
if v == 0:
|
||||
raise ValueError("amount must be non-zero")
|
||||
return v
|
||||
|
||||
+9
-23
@@ -1,33 +1,19 @@
|
||||
from dataclasses import dataclass, field
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
@dataclass
|
||||
class ForecastScenario:
|
||||
id: UUID = field(default_factory=uuid4)
|
||||
|
||||
class ForecastScenario(BaseModel):
|
||||
id: UUID = Field(default_factory=uuid4)
|
||||
name: str = "baseline"
|
||||
income_multiplier: float = 1.0
|
||||
expense_multiplier: float = 1.0
|
||||
growth_multiplier: float = 1.0
|
||||
description: str = ""
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": str(self.id),
|
||||
"name": self.name,
|
||||
"income_multiplier": self.income_multiplier,
|
||||
"expense_multiplier": self.expense_multiplier,
|
||||
"growth_multiplier": self.growth_multiplier,
|
||||
"description": self.description,
|
||||
}
|
||||
|
||||
@field_validator("income_multiplier", "expense_multiplier", "growth_multiplier")
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ForecastScenario":
|
||||
return cls(
|
||||
id=UUID(data["id"]),
|
||||
name=data["name"],
|
||||
income_multiplier=data.get("income_multiplier", 1.0),
|
||||
expense_multiplier=data.get("expense_multiplier", 1.0),
|
||||
growth_multiplier=data.get("growth_multiplier", 1.0),
|
||||
description=data.get("description", ""),
|
||||
)
|
||||
def _non_negative(cls, v: float) -> float:
|
||||
if v < 0:
|
||||
raise ValueError("multiplier must be non-negative")
|
||||
return v
|
||||
|
||||
+9
-23
@@ -1,33 +1,19 @@
|
||||
from dataclasses import dataclass, field
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
@dataclass
|
||||
class Transaction:
|
||||
id: UUID = field(default_factory=uuid4)
|
||||
|
||||
class Transaction(BaseModel):
|
||||
id: UUID = Field(default_factory=uuid4)
|
||||
date: str = ""
|
||||
account: str = ""
|
||||
category: str = ""
|
||||
amount: float = 0.0
|
||||
description: str = ""
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": str(self.id),
|
||||
"date": self.date,
|
||||
"account": self.account,
|
||||
"category": self.category,
|
||||
"amount": self.amount,
|
||||
"description": self.description,
|
||||
}
|
||||
|
||||
@field_validator("amount")
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Transaction":
|
||||
return cls(
|
||||
id=UUID(data["id"]),
|
||||
date=data["date"],
|
||||
account=data.get("account", ""),
|
||||
category=data.get("category", ""),
|
||||
amount=data.get("amount", 0.0),
|
||||
description=data.get("description", ""),
|
||||
)
|
||||
def _amount_nonzero(cls, v: float) -> float:
|
||||
if v == 0:
|
||||
raise ValueError("amount must be non-zero")
|
||||
return v
|
||||
|
||||
@@ -119,17 +119,19 @@ class ExcelSync:
|
||||
|
||||
def _add_to_model(self, model: FinancialModel, sheet_name: str, data: dict) -> None:
|
||||
try:
|
||||
cls = _SHEET_CONFIG[sheet_name]["cls"]
|
||||
instance = cls.model_validate(data)
|
||||
if sheet_name == "Accounts":
|
||||
model.accounts.append(Account.from_dict(data))
|
||||
model.accounts.append(instance)
|
||||
elif sheet_name == "Transactions":
|
||||
model.transactions.append(Transaction.from_dict(data))
|
||||
model.transactions.append(instance)
|
||||
elif sheet_name == "Recurring":
|
||||
model.recurring.append(RecurringCashflow.from_dict(data))
|
||||
model.recurring.append(instance)
|
||||
elif sheet_name == "Assets":
|
||||
model.assets.append(Asset.from_dict(data))
|
||||
model.assets.append(instance)
|
||||
elif sheet_name == "Liabilities":
|
||||
model.liabilities.append(Liability.from_dict(data))
|
||||
model.liabilities.append(instance)
|
||||
elif sheet_name == "ExchangeRates":
|
||||
model.exchange_rates.append(ExchangeRate.from_dict(data))
|
||||
model.exchange_rates.append(instance)
|
||||
except Exception as e:
|
||||
raise SyncError(f"Failed to parse row in {sheet_name}: {e}") from e
|
||||
|
||||
@@ -11,6 +11,7 @@ dependencies = [
|
||||
"openpyxl>=3.1",
|
||||
"typer>=0.9",
|
||||
"rich>=13.0",
|
||||
"pydantic>=2.0",
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
|
||||
@@ -6,8 +6,8 @@ from domain import CurrencyConverter, CurrencyError, ExchangeRate
|
||||
class TestExchangeRate:
|
||||
def test_to_dict_roundtrip(self):
|
||||
rate = ExchangeRate(from_currency="USD", to_currency="RUB", rate=80.0)
|
||||
d = rate.to_dict()
|
||||
r2 = ExchangeRate.from_dict(d)
|
||||
d = rate.model_dump(mode="json")
|
||||
r2 = ExchangeRate.model_validate(d)
|
||||
assert r2.from_currency == "USD"
|
||||
assert r2.to_currency == "RUB"
|
||||
assert r2.rate == 80.0
|
||||
|
||||
+4
-4
@@ -21,8 +21,8 @@ class TestAccount:
|
||||
|
||||
def test_to_dict_roundtrip(self):
|
||||
a = Account(name="Test", balance=100.0)
|
||||
d = a.to_dict()
|
||||
a2 = Account.from_dict(d)
|
||||
d = a.model_dump(mode="json")
|
||||
a2 = Account.model_validate(d)
|
||||
assert a2.name == a.name
|
||||
assert a2.balance == a.balance
|
||||
assert a2.currency == a.currency
|
||||
@@ -35,8 +35,8 @@ class TestTransaction:
|
||||
|
||||
def test_roundtrip(self):
|
||||
t = Transaction(amount=-100.0, category="rent", description="test")
|
||||
d = t.to_dict()
|
||||
t2 = Transaction.from_dict(d)
|
||||
d = t.model_dump(mode="json")
|
||||
t2 = Transaction.model_validate(d)
|
||||
assert t2.amount == t.amount
|
||||
assert t2.category == t.category
|
||||
assert t2.description == t.description
|
||||
|
||||
Reference in New Issue
Block a user