zwitschi
97b1c0360b
- Updated test functions in various test files to enhance code clarity by formatting long lines and improving indentation. - Adjusted assertions to use multi-line formatting for better readability. - Added new test cases for theme settings API to ensure proper functionality. - Ensured consistent use of line breaks and spacing across test files for uniformity.
72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
from sqlalchemy import event, text
|
|
from sqlalchemy import Column, Integer, Float, String, ForeignKey
|
|
from sqlalchemy.orm import relationship
|
|
from config.database import Base
|
|
|
|
|
|
class Capex(Base):
|
|
__tablename__ = "capex"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
scenario_id = Column(Integer, ForeignKey("scenario.id"), nullable=False)
|
|
amount = Column(Float, nullable=False)
|
|
description = Column(String, nullable=True)
|
|
currency_id = Column(Integer, ForeignKey("currency.id"), nullable=False)
|
|
|
|
scenario = relationship("Scenario", back_populates="capex_items")
|
|
currency = relationship("Currency", back_populates="capex_items")
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f"<Capex id={self.id} scenario_id={self.scenario_id} "
|
|
f"amount={self.amount} currency_id={self.currency_id}>"
|
|
)
|
|
|
|
@property
|
|
def currency_code(self) -> str:
|
|
return self.currency.code if self.currency else None
|
|
|
|
@currency_code.setter
|
|
def currency_code(self, value: str) -> None:
|
|
# store pending code so application code or migrations can pick it up
|
|
setattr(
|
|
self, "_currency_code_pending", (value or "USD").strip().upper()
|
|
)
|
|
|
|
|
|
# SQLAlchemy event handlers to ensure currency_id is set before insert/update
|
|
|
|
|
|
def _resolve_currency(mapper, connection, target):
|
|
# If currency_id already set, nothing to do
|
|
if getattr(target, "currency_id", None):
|
|
return
|
|
code = getattr(target, "_currency_code_pending", None) or "USD"
|
|
# Try to find existing currency id
|
|
row = connection.execute(
|
|
text("SELECT id FROM currency WHERE code = :code"), {"code": code}
|
|
).fetchone()
|
|
if row:
|
|
cid = row[0]
|
|
else:
|
|
# Insert new currency and attempt to get lastrowid
|
|
res = connection.execute(
|
|
text(
|
|
"INSERT INTO currency (code, name, symbol, is_active) VALUES (:code, :name, :symbol, :active)"
|
|
),
|
|
{"code": code, "name": code, "symbol": None, "active": True},
|
|
)
|
|
try:
|
|
cid = res.lastrowid
|
|
except Exception:
|
|
# fallback: select after insert
|
|
cid = connection.execute(
|
|
text("SELECT id FROM currency WHERE code = :code"),
|
|
{"code": code},
|
|
).scalar()
|
|
target.currency_id = cid
|
|
|
|
|
|
event.listen(Capex, "before_insert", _resolve_currency)
|
|
event.listen(Capex, "before_update", _resolve_currency)
|