From c765f3672a4b68d2339f22c9f2aca7673de56b88 Mon Sep 17 00:00:00 2001 From: oqyude Date: Thu, 8 Oct 2026 16:42:04 +0300 Subject: [PATCH] =?UTF-8?q?T2:=20Pydantic=20v2=20=E2=80=94=20replace=20@da?= =?UTF-8?q?taclass=20with=20BaseModel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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. --- domain/account.py | 28 +++++--------- domain/asset.py | 25 ++---------- domain/currency.py | 23 ++++------- domain/liability.py | 30 +++++---------- domain/model.py | 63 +++++++++++++++++-------------- domain/recurring.py | 39 ++++++++----------- domain/scenario.py | 32 +++++----------- domain/transaction.py | 32 +++++----------- infrastructure/sync/excel_sync.py | 14 ++++--- pyproject.toml | 1 + tests/test_currency.py | 4 +- tests/test_model.py | 8 ++-- 12 files changed, 112 insertions(+), 187 deletions(-) diff --git a/domain/account.py b/domain/account.py index f70ca25..3949112 100644 --- a/domain/account.py +++ b/domain/account.py @@ -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 diff --git a/domain/asset.py b/domain/asset.py index e68f670..c2bad2a 100644 --- a/domain/asset.py +++ b/domain/asset.py @@ -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), - ) diff --git a/domain/currency.py b/domain/currency.py index 1a0ea53..e7baf0a 100644 --- a/domain/currency.py +++ b/domain/currency.py @@ -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] = [ diff --git a/domain/liability.py b/domain/liability.py index bdbaa01..8a2e74c 100644 --- a/domain/liability.py +++ b/domain/liability.py @@ -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 diff --git a/domain/model.py b/domain/model.py index 17c61ed..ec49339 100644 --- a/domain/model.py +++ b/domain/model.py @@ -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: diff --git a/domain/recurring.py b/domain/recurring.py index 6ca69f4..4daec87 100644 --- a/domain/recurring.py +++ b/domain/recurring.py @@ -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 diff --git a/domain/scenario.py b/domain/scenario.py index 349c5ba..0dd33b6 100644 --- a/domain/scenario.py +++ b/domain/scenario.py @@ -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 diff --git a/domain/transaction.py b/domain/transaction.py index 0039556..3a4e4ae 100644 --- a/domain/transaction.py +++ b/domain/transaction.py @@ -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 diff --git a/infrastructure/sync/excel_sync.py b/infrastructure/sync/excel_sync.py index b7d542c..953dc1f 100644 --- a/infrastructure/sync/excel_sync.py +++ b/infrastructure/sync/excel_sync.py @@ -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 diff --git a/pyproject.toml b/pyproject.toml index fb257c9..a2a9c9e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,7 @@ dependencies = [ "openpyxl>=3.1", "typer>=0.9", "rich>=13.0", + "pydantic>=2.0", ] [project.scripts] diff --git a/tests/test_currency.py b/tests/test_currency.py index 877d0e1..76bc7cf 100644 --- a/tests/test_currency.py +++ b/tests/test_currency.py @@ -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 diff --git a/tests/test_model.py b/tests/test_model.py index 8442829..94880f3 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -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