Files
calminer/routes/scenarios.py
zwitschi 522b1e4105
Some checks failed
CI / test (push) Has been skipped
CI / build (push) Has been skipped
CI / lint (push) Failing after 15s
CI / deploy (push) Has been skipped
feat: add scenarios list page with metrics and quick actions
- Introduced a new template for listing scenarios associated with a project.
- Added metrics for total, active, draft, and archived scenarios.
- Implemented quick actions for creating new scenarios and reviewing project overview.
- Enhanced navigation with breadcrumbs for better user experience.

refactor: update Opex and Profitability templates for consistency

- Changed titles and button labels for clarity in Opex and Profitability templates.
- Updated form IDs and action URLs for better alignment with new naming conventions.
- Improved navigation links to include scenario and project overviews.

test: add integration tests for Opex calculations

- Created new tests for Opex calculation HTML and JSON flows.
- Validated successful calculations and ensured correct data persistence.
- Implemented tests for currency mismatch and unsupported frequency scenarios.

test: enhance project and scenario route tests

- Added tests to verify scenario list rendering and calculator shortcuts.
- Ensured scenario detail pages link back to the portfolio correctly.
- Validated project detail pages show associated scenarios accurately.
2025-11-13 16:21:36 +01:00

657 lines
20 KiB
Python

from __future__ import annotations
from datetime import date
from types import SimpleNamespace
from typing import List
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
from fastapi.responses import HTMLResponse, RedirectResponse
from dependencies import (
get_pricing_metadata,
get_unit_of_work,
require_any_role,
require_any_role_html,
require_roles,
require_roles_html,
require_scenario_resource,
require_scenario_resource_html,
)
from models import ResourceType, Scenario, ScenarioStatus, User
from schemas.scenario import (
ScenarioComparisonRequest,
ScenarioComparisonResponse,
ScenarioCreate,
ScenarioRead,
ScenarioUpdate,
)
from services.currency import CurrencyValidationError, normalise_currency
from services.exceptions import (
EntityConflictError,
EntityNotFoundError,
ScenarioValidationError,
)
from services.pricing import PricingMetadata
from services.unit_of_work import UnitOfWork
from routes.template_filters import create_templates
router = APIRouter(tags=["Scenarios"])
templates = create_templates()
READ_ROLES = ("viewer", "analyst", "project_manager", "admin")
MANAGE_ROLES = ("project_manager", "admin")
def _to_read_model(scenario: Scenario) -> ScenarioRead:
return ScenarioRead.model_validate(scenario)
def _resource_type_choices() -> list[tuple[str, str]]:
return [
(resource.value, resource.value.replace("_", " ").title())
for resource in ResourceType
]
def _scenario_status_choices() -> list[tuple[str, str]]:
return [
(status.value, status.value.title()) for status in ScenarioStatus
]
def _require_project_repo(uow: UnitOfWork):
if not uow.projects:
raise RuntimeError("Project repository not initialised")
return uow.projects
def _require_scenario_repo(uow: UnitOfWork):
if not uow.scenarios:
raise RuntimeError("Scenario repository not initialised")
return uow.scenarios
@router.get(
"/projects/{project_id}/scenarios",
response_model=List[ScenarioRead],
)
def list_scenarios_for_project(
project_id: int,
_: User = Depends(require_any_role(*READ_ROLES)),
uow: UnitOfWork = Depends(get_unit_of_work),
) -> List[ScenarioRead]:
project_repo = _require_project_repo(uow)
scenario_repo = _require_scenario_repo(uow)
try:
project_repo.get(project_id)
except EntityNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
scenarios = scenario_repo.list_for_project(project_id)
return [_to_read_model(scenario) for scenario in scenarios]
@router.post(
"/projects/{project_id}/scenarios/compare",
response_model=ScenarioComparisonResponse,
status_code=status.HTTP_200_OK,
)
def compare_scenarios(
project_id: int,
payload: ScenarioComparisonRequest,
_: User = Depends(require_any_role(*READ_ROLES)),
uow: UnitOfWork = Depends(get_unit_of_work),
) -> ScenarioComparisonResponse:
try:
_require_project_repo(uow).get(project_id)
except EntityNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
) from exc
try:
scenarios = uow.validate_scenarios_for_comparison(payload.scenario_ids)
if any(scenario.project_id != project_id for scenario in scenarios):
raise ScenarioValidationError(
code="SCENARIO_PROJECT_MISMATCH",
message="Selected scenarios do not belong to the same project.",
scenario_ids=[
scenario.id for scenario in scenarios if scenario.id is not None
],
)
except EntityNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
) from exc
except ScenarioValidationError as exc:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail={
"code": exc.code,
"message": exc.message,
"scenario_ids": list(exc.scenario_ids or []),
},
) from exc
return ScenarioComparisonResponse(
project_id=project_id,
scenarios=[_to_read_model(scenario) for scenario in scenarios],
)
@router.post(
"/projects/{project_id}/scenarios",
response_model=ScenarioRead,
status_code=status.HTTP_201_CREATED,
)
def create_scenario_for_project(
project_id: int,
payload: ScenarioCreate,
_: User = Depends(require_roles(*MANAGE_ROLES)),
uow: UnitOfWork = Depends(get_unit_of_work),
metadata: PricingMetadata = Depends(get_pricing_metadata),
) -> ScenarioRead:
project_repo = _require_project_repo(uow)
scenario_repo = _require_scenario_repo(uow)
try:
project_repo.get(project_id)
except EntityNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
scenario_data = payload.model_dump()
if not scenario_data.get("currency") and metadata.default_currency:
scenario_data["currency"] = metadata.default_currency
scenario = Scenario(project_id=project_id, **scenario_data)
try:
created = scenario_repo.create(scenario)
except EntityConflictError as exc:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
return _to_read_model(created)
@router.get(
"/projects/{project_id}/scenarios/ui",
response_class=HTMLResponse,
include_in_schema=False,
name="scenarios.project_scenario_list",
)
def project_scenario_list_page(
project_id: int,
request: Request,
_: User = Depends(require_any_role_html(*READ_ROLES)),
uow: UnitOfWork = Depends(get_unit_of_work),
) -> HTMLResponse:
try:
project = _require_project_repo(uow).get(
project_id, with_children=True)
except EntityNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
) from exc
scenarios = sorted(
project.scenarios,
key=lambda scenario: scenario.updated_at or scenario.created_at,
reverse=True,
)
scenario_totals = {
"total": len(scenarios),
"active": sum(
1 for scenario in scenarios if scenario.status == ScenarioStatus.ACTIVE
),
"draft": sum(
1 for scenario in scenarios if scenario.status == ScenarioStatus.DRAFT
),
"archived": sum(
1 for scenario in scenarios if scenario.status == ScenarioStatus.ARCHIVED
),
"latest_update": max(
(
scenario.updated_at or scenario.created_at
for scenario in scenarios
if scenario.updated_at or scenario.created_at
),
default=None,
),
}
return templates.TemplateResponse(
request,
"scenarios/list.html",
{
"project": project,
"scenarios": scenarios,
"scenario_totals": scenario_totals,
},
)
@router.get("/scenarios/{scenario_id}", response_model=ScenarioRead)
def get_scenario(
scenario: Scenario = Depends(require_scenario_resource()),
) -> ScenarioRead:
return _to_read_model(scenario)
@router.put("/scenarios/{scenario_id}", response_model=ScenarioRead)
def update_scenario(
payload: ScenarioUpdate,
scenario: Scenario = Depends(
require_scenario_resource(require_manage=True)
),
uow: UnitOfWork = Depends(get_unit_of_work),
) -> ScenarioRead:
update_data = payload.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(scenario, field, value)
uow.flush()
return _to_read_model(scenario)
@router.delete("/scenarios/{scenario_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_scenario(
scenario: Scenario = Depends(
require_scenario_resource(require_manage=True)
),
uow: UnitOfWork = Depends(get_unit_of_work),
) -> None:
_require_scenario_repo(uow).delete(scenario.id)
def _normalise(value: str | None) -> str | None:
if value is None:
return None
value = value.strip()
return value or None
def _parse_date(value: str | None) -> date | None:
value = _normalise(value)
if not value:
return None
return date.fromisoformat(value)
def _parse_discount_rate(value: str | None) -> float | None:
value = _normalise(value)
if not value:
return None
try:
return float(value)
except ValueError:
return None
def _scenario_form_state(
*,
project_id: int,
name: str,
description: str | None,
status: ScenarioStatus,
start_date: date | None,
end_date: date | None,
discount_rate: float | None,
currency: str | None,
primary_resource: ResourceType | None,
scenario_id: int | None = None,
) -> SimpleNamespace:
return SimpleNamespace(
id=scenario_id,
project_id=project_id,
name=name,
description=description,
status=status,
start_date=start_date,
end_date=end_date,
discount_rate=discount_rate,
currency=currency,
primary_resource=primary_resource,
)
@router.get(
"/projects/{project_id}/scenarios/new",
response_class=HTMLResponse,
include_in_schema=False,
name="scenarios.create_scenario_form",
)
def create_scenario_form(
project_id: int,
request: Request,
_: User = Depends(require_roles_html(*MANAGE_ROLES)),
uow: UnitOfWork = Depends(get_unit_of_work),
metadata: PricingMetadata = Depends(get_pricing_metadata),
) -> HTMLResponse:
try:
project = _require_project_repo(uow).get(project_id)
except EntityNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
) from exc
return templates.TemplateResponse(
request,
"scenarios/form.html",
{
"project": project,
"scenario": None,
"scenario_statuses": _scenario_status_choices(),
"resource_types": _resource_type_choices(),
"form_action": request.url_for(
"scenarios.create_scenario_submit", project_id=project_id
),
"cancel_url": request.url_for(
"projects.view_project", project_id=project_id
),
"default_currency": metadata.default_currency,
},
)
@router.post(
"/projects/{project_id}/scenarios/new",
include_in_schema=False,
name="scenarios.create_scenario_submit",
)
def create_scenario_submit(
project_id: int,
request: Request,
_: User = Depends(require_roles_html(*MANAGE_ROLES)),
name: str = Form(...),
description: str | None = Form(None),
status_value: str = Form(ScenarioStatus.DRAFT.value),
start_date: str | None = Form(None),
end_date: str | None = Form(None),
discount_rate: str | None = Form(None),
currency: str | None = Form(None),
primary_resource: str | None = Form(None),
uow: UnitOfWork = Depends(get_unit_of_work),
metadata: PricingMetadata = Depends(get_pricing_metadata),
):
project_repo = _require_project_repo(uow)
scenario_repo = _require_scenario_repo(uow)
try:
project = project_repo.get(project_id)
except EntityNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
) from exc
try:
status_enum = ScenarioStatus(status_value)
except ValueError:
status_enum = ScenarioStatus.DRAFT
resource_enum = None
if primary_resource:
try:
resource_enum = ResourceType(primary_resource)
except ValueError:
resource_enum = None
name_value = name.strip()
description_value = _normalise(description)
start_date_value = _parse_date(start_date)
end_date_value = _parse_date(end_date)
discount_rate_value = _parse_discount_rate(discount_rate)
currency_input = _normalise(currency)
effective_currency = currency_input or metadata.default_currency
try:
currency_value = (
normalise_currency(effective_currency)
if effective_currency else None
)
except CurrencyValidationError as exc:
form_state = _scenario_form_state(
project_id=project_id,
name=name_value,
description=description_value,
status=status_enum,
start_date=start_date_value,
end_date=end_date_value,
discount_rate=discount_rate_value,
currency=currency_input or metadata.default_currency,
primary_resource=resource_enum,
)
return templates.TemplateResponse(
request,
"scenarios/form.html",
{
"project": project,
"scenario": form_state,
"scenario_statuses": _scenario_status_choices(),
"resource_types": _resource_type_choices(),
"form_action": request.url_for(
"scenarios.create_scenario_submit", project_id=project_id
),
"cancel_url": request.url_for(
"projects.view_project", project_id=project_id
),
"error": str(exc),
"error_field": "currency",
"default_currency": metadata.default_currency,
},
status_code=status.HTTP_400_BAD_REQUEST,
)
scenario = Scenario(
project_id=project_id,
name=name_value,
description=description_value,
status=status_enum,
start_date=start_date_value,
end_date=end_date_value,
discount_rate=discount_rate_value,
currency=currency_value,
primary_resource=resource_enum,
)
try:
scenario_repo.create(scenario)
except EntityConflictError:
return templates.TemplateResponse(
request,
"scenarios/form.html",
{
"project": project,
"scenario": scenario,
"scenario_statuses": _scenario_status_choices(),
"resource_types": _resource_type_choices(),
"form_action": request.url_for(
"scenarios.create_scenario_submit", project_id=project_id
),
"cancel_url": request.url_for(
"projects.view_project", project_id=project_id
),
"error": "Scenario with this name already exists for this project.",
"error_field": "name",
"default_currency": metadata.default_currency,
},
status_code=status.HTTP_409_CONFLICT,
)
return RedirectResponse(
request.url_for("projects.view_project", project_id=project_id),
status_code=status.HTTP_303_SEE_OTHER,
)
@router.get(
"/scenarios/{scenario_id}/view",
response_class=HTMLResponse,
include_in_schema=False,
name="scenarios.view_scenario",
)
def view_scenario(
request: Request,
_: User = Depends(require_any_role_html(*READ_ROLES)),
scenario: Scenario = Depends(
require_scenario_resource_html(with_children=True)
),
uow: UnitOfWork = Depends(get_unit_of_work),
) -> HTMLResponse:
project = _require_project_repo(uow).get(scenario.project_id)
financial_inputs = sorted(
scenario.financial_inputs, key=lambda item: item.created_at
)
simulation_parameters = sorted(
scenario.simulation_parameters, key=lambda item: item.created_at
)
scenario_metrics = {
"financial_count": len(financial_inputs),
"parameter_count": len(simulation_parameters),
"currency": scenario.currency,
"primary_resource": scenario.primary_resource.value.replace('_', ' ').title() if scenario.primary_resource else None,
}
return templates.TemplateResponse(
request,
"scenarios/detail.html",
{
"project": project,
"scenario": scenario,
"scenario_metrics": scenario_metrics,
"financial_inputs": financial_inputs,
"simulation_parameters": simulation_parameters,
},
)
@router.get(
"/scenarios/{scenario_id}/edit",
response_class=HTMLResponse,
include_in_schema=False,
name="scenarios.edit_scenario_form",
)
def edit_scenario_form(
request: Request,
_: User = Depends(require_roles_html(*MANAGE_ROLES)),
scenario: Scenario = Depends(
require_scenario_resource_html(require_manage=True)
),
uow: UnitOfWork = Depends(get_unit_of_work),
metadata: PricingMetadata = Depends(get_pricing_metadata),
) -> HTMLResponse:
project = _require_project_repo(uow).get(scenario.project_id)
return templates.TemplateResponse(
request,
"scenarios/form.html",
{
"project": project,
"scenario": scenario,
"scenario_statuses": _scenario_status_choices(),
"resource_types": _resource_type_choices(),
"form_action": request.url_for(
"scenarios.edit_scenario_submit", scenario_id=scenario.id
),
"cancel_url": request.url_for(
"scenarios.view_scenario", scenario_id=scenario.id
),
"default_currency": metadata.default_currency,
},
)
@router.post(
"/scenarios/{scenario_id}/edit",
include_in_schema=False,
name="scenarios.edit_scenario_submit",
)
def edit_scenario_submit(
request: Request,
_: User = Depends(require_roles_html(*MANAGE_ROLES)),
scenario: Scenario = Depends(
require_scenario_resource_html(require_manage=True)
),
name: str = Form(...),
description: str | None = Form(None),
status_value: str = Form(ScenarioStatus.DRAFT.value),
start_date: str | None = Form(None),
end_date: str | None = Form(None),
discount_rate: str | None = Form(None),
currency: str | None = Form(None),
primary_resource: str | None = Form(None),
uow: UnitOfWork = Depends(get_unit_of_work),
metadata: PricingMetadata = Depends(get_pricing_metadata),
):
project = _require_project_repo(uow).get(scenario.project_id)
name_value = name.strip()
description_value = _normalise(description)
try:
scenario.status = ScenarioStatus(status_value)
except ValueError:
scenario.status = ScenarioStatus.DRAFT
status_enum = scenario.status
resource_enum = None
if primary_resource:
try:
resource_enum = ResourceType(primary_resource)
except ValueError:
resource_enum = None
start_date_value = _parse_date(start_date)
end_date_value = _parse_date(end_date)
discount_rate_value = _parse_discount_rate(discount_rate)
currency_input = _normalise(currency)
try:
currency_value = normalise_currency(currency_input)
except CurrencyValidationError as exc:
form_state = _scenario_form_state(
scenario_id=scenario.id,
project_id=scenario.project_id,
name=name_value,
description=description_value,
status=status_enum,
start_date=start_date_value,
end_date=end_date_value,
discount_rate=discount_rate_value,
currency=currency_input,
primary_resource=resource_enum,
)
return templates.TemplateResponse(
request,
"scenarios/form.html",
{
"project": project,
"scenario": form_state,
"scenario_statuses": _scenario_status_choices(),
"resource_types": _resource_type_choices(),
"form_action": request.url_for(
"scenarios.edit_scenario_submit", scenario_id=scenario.id
),
"cancel_url": request.url_for(
"scenarios.view_scenario", scenario_id=scenario.id
),
"error": str(exc),
"error_field": "currency",
"default_currency": metadata.default_currency,
},
status_code=status.HTTP_400_BAD_REQUEST,
)
scenario.name = name_value
scenario.description = description_value
scenario.start_date = start_date_value
scenario.end_date = end_date_value
scenario.discount_rate = discount_rate_value
scenario.currency = currency_value
scenario.primary_resource = resource_enum
uow.flush()
return RedirectResponse(
request.url_for("scenarios.view_scenario", scenario_id=scenario.id),
status_code=status.HTTP_303_SEE_OTHER,
)