118 lines
4.0 KiB
Python
118 lines
4.0 KiB
Python
"""Position CRUD endpoints."""
|
|
|
|
import math
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlmodel import Session, select
|
|
|
|
from ten31portal.audit import record_audit
|
|
from ten31portal.auth import get_current_user, require_writer
|
|
from ten31portal.database import get_session
|
|
from ten31portal.models import Holding, Position, Valuation, ValuationRound, RoundStatus, User
|
|
from ten31portal.schemas import PositionCreate, PositionResponse, PositionUpdate
|
|
|
|
router = APIRouter(tags=["positions"])
|
|
|
|
|
|
def _dollars_to_cents(dollars: float) -> int:
|
|
"""Convert dollar amount to integer cents, rounding to nearest cent."""
|
|
return round(dollars * 100)
|
|
|
|
|
|
@router.get("/api/holdings/{holding_id}/positions")
|
|
def list_positions(
|
|
holding_id: int,
|
|
user: User = Depends(get_current_user),
|
|
session: Session = Depends(get_session),
|
|
) -> list[PositionResponse]:
|
|
holding = session.get(Holding, holding_id)
|
|
if holding is None:
|
|
raise HTTPException(status_code=404, detail="Holding not found")
|
|
rows = session.exec(select(Position).where(Position.holding_id == holding_id)).all()
|
|
return [PositionResponse.model_validate(r, from_attributes=True) for r in rows]
|
|
|
|
|
|
@router.post("/api/holdings/{holding_id}/positions", status_code=201)
|
|
def create_position(
|
|
holding_id: int,
|
|
body: PositionCreate,
|
|
user: User = Depends(require_writer),
|
|
session: Session = Depends(get_session),
|
|
) -> PositionResponse:
|
|
holding = session.get(Holding, holding_id)
|
|
if holding is None:
|
|
raise HTTPException(status_code=404, detail="Holding not found")
|
|
position = Position(
|
|
holding_id=holding_id,
|
|
security_name=body.security_name,
|
|
investment_date=body.investment_date,
|
|
shares=body.shares,
|
|
cost_cents=_dollars_to_cents(body.cost_dollars),
|
|
)
|
|
session.add(position)
|
|
session.flush()
|
|
record_audit(session, user.id, "create", "position", position.id, {
|
|
"holding_id": holding_id,
|
|
"security_name": body.security_name,
|
|
"cost_dollars": body.cost_dollars,
|
|
})
|
|
session.commit()
|
|
session.refresh(position)
|
|
return PositionResponse.model_validate(position, from_attributes=True)
|
|
|
|
|
|
@router.patch("/api/positions/{position_id}")
|
|
def update_position(
|
|
position_id: int,
|
|
body: PositionUpdate,
|
|
user: User = Depends(require_writer),
|
|
session: Session = Depends(get_session),
|
|
) -> PositionResponse:
|
|
position = session.get(Position, position_id)
|
|
if position is None:
|
|
raise HTTPException(status_code=404, detail="Position not found")
|
|
changes = body.model_dump(exclude_unset=True)
|
|
if "cost_dollars" in changes:
|
|
position.cost_cents = _dollars_to_cents(changes.pop("cost_dollars"))
|
|
for key, val in changes.items():
|
|
setattr(position, key, val)
|
|
session.add(position)
|
|
session.flush()
|
|
record_audit(session, user.id, "update", "position", position.id, body.model_dump(exclude_unset=True))
|
|
session.commit()
|
|
session.refresh(position)
|
|
return PositionResponse.model_validate(position, from_attributes=True)
|
|
|
|
|
|
@router.delete("/api/positions/{position_id}")
|
|
def delete_position(
|
|
position_id: int,
|
|
user: User = Depends(require_writer),
|
|
session: Session = Depends(get_session),
|
|
) -> dict[str, str]:
|
|
position = session.get(Position, position_id)
|
|
if position is None:
|
|
raise HTTPException(status_code=404, detail="Position not found")
|
|
|
|
# Refuse if any approved valuation references this position
|
|
approved_vals = session.exec(
|
|
select(Valuation)
|
|
.join(ValuationRound)
|
|
.where(
|
|
Valuation.position_id == position_id,
|
|
ValuationRound.status == RoundStatus.approved,
|
|
)
|
|
).all()
|
|
if approved_vals:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="Cannot delete position referenced by an approved valuation round.",
|
|
)
|
|
|
|
record_audit(session, user.id, "delete", "position", position.id, {
|
|
"security_name": position.security_name,
|
|
})
|
|
session.delete(position)
|
|
session.commit()
|
|
return {"status": "deleted"}
|