feat: add initial database models and changelog for financial inputs and projects

This commit is contained in:
2025-11-09 16:50:14 +01:00
parent c6a0eb2588
commit 203a5d08f2
7 changed files with 282 additions and 30 deletions

5
changelog.md Normal file
View File

@@ -0,0 +1,5 @@
# Changelog
## 2025-11-09
- Captured current implementation status, requirements coverage, missing features, and prioritized roadmap in `calminer-docs/implementation_status.md` to guide future development.

37
main.py
View File

@@ -1,25 +1,17 @@
from routes.distributions import router as distributions_router
from routes.ui import router as ui_router
from routes.parameters import router as parameters_router
from typing import Awaitable, Callable
from fastapi import FastAPI, Request, Response
from fastapi.staticfiles import StaticFiles
from middleware.validation import validate_json
from config.database import Base, engine
from routes.scenarios import router as scenarios_router
from routes.costs import router as costs_router
from routes.consumption import router as consumption_router
from routes.production import router as production_router
from routes.equipment import router as equipment_router
from routes.reporting import router as reporting_router
from routes.currencies import router as currencies_router
from routes.simulations import router as simulations_router
from routes.maintenance import router as maintenance_router
from routes.settings import router as settings_router
from routes.users import router as users_router
from models import (
FinancialInput,
Project,
Scenario,
SimulationParameter,
)
# Initialize database schema
# Initialize database schema (imports above ensure models are registered)
Base.metadata.create_all(bind=engine)
app = FastAPI()
@@ -39,18 +31,3 @@ async def health() -> dict[str, str]:
app.mount("/static", StaticFiles(directory="static"), name="static")
# Include API routers
app.include_router(scenarios_router)
app.include_router(parameters_router)
app.include_router(distributions_router)
app.include_router(costs_router)
app.include_router(consumption_router)
app.include_router(simulations_router)
app.include_router(production_router)
app.include_router(equipment_router)
app.include_router(maintenance_router)
app.include_router(reporting_router)
app.include_router(currencies_router)
app.include_router(settings_router)
app.include_router(ui_router)
app.include_router(users_router)

17
models/__init__.py Normal file
View File

@@ -0,0 +1,17 @@
"""Database models for the CalMiner domain."""
from .financial_input import FinancialCategory, FinancialInput
from .project import MiningOperationType, Project
from .scenario import Scenario, ScenarioStatus
from .simulation_parameter import DistributionType, SimulationParameter
__all__ = [
"FinancialCategory",
"FinancialInput",
"MiningOperationType",
"Project",
"Scenario",
"ScenarioStatus",
"DistributionType",
"SimulationParameter",
]

63
models/financial_input.py Normal file
View File

@@ -0,0 +1,63 @@
from __future__ import annotations
from datetime import date, datetime
from enum import Enum
from typing import TYPE_CHECKING
from sqlalchemy import (
Date,
DateTime,
Enum as SQLEnum,
ForeignKey,
Integer,
Numeric,
String,
Text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from config.database import Base
if TYPE_CHECKING: # pragma: no cover
from .scenario import Scenario
class FinancialCategory(str, Enum):
"""Enumeration of cost and revenue classifications."""
CAPITAL_EXPENDITURE = "capex"
OPERATING_EXPENDITURE = "opex"
REVENUE = "revenue"
CONTINGENCY = "contingency"
OTHER = "other"
class FinancialInput(Base):
"""Line-item financial assumption attached to a scenario."""
__tablename__ = "financial_inputs"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
scenario_id: Mapped[int] = mapped_column(
ForeignKey("scenarios.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(255), nullable=False)
category: Mapped[FinancialCategory] = mapped_column(
SQLEnum(FinancialCategory), nullable=False
)
amount: Mapped[float] = mapped_column(Numeric(18, 2), nullable=False)
currency: Mapped[str | None] = mapped_column(String(3), nullable=True)
effective_date: Mapped[date | None] = mapped_column(Date, nullable=True)
notes: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
)
scenario: Mapped["Scenario"] = relationship("Scenario", back_populates="financial_inputs")
def __repr__(self) -> str: # pragma: no cover
return f"FinancialInput(id={self.id!r}, scenario_id={self.scenario_id!r}, name={self.name!r})"

54
models/project.py Normal file
View File

@@ -0,0 +1,54 @@
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import TYPE_CHECKING, List
from sqlalchemy import DateTime, Enum as SQLEnum, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from config.database import Base
if TYPE_CHECKING: # pragma: no cover
from .scenario import Scenario
class MiningOperationType(str, Enum):
"""Supported mining operation categories."""
OPEN_PIT = "open_pit"
UNDERGROUND = "underground"
PLACER = "placer"
QUARRY = "quarry"
OTHER = "other"
class Project(Base):
"""Top-level mining project grouping multiple scenarios."""
__tablename__ = "projects"
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
location: Mapped[str | None] = mapped_column(String(255), nullable=True)
operation_type: Mapped[MiningOperationType] = mapped_column(
SQLEnum(MiningOperationType), nullable=False, default=MiningOperationType.OTHER
)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
)
scenarios: Mapped[List["Scenario"]] = relationship(
"Scenario",
back_populates="project",
cascade="all, delete-orphan",
passive_deletes=True,
)
def __repr__(self) -> str: # pragma: no cover - helpful for debugging
return f"Project(id={self.id!r}, name={self.name!r})"

67
models/scenario.py Normal file
View File

@@ -0,0 +1,67 @@
from __future__ import annotations
from datetime import date, datetime
from enum import Enum
from typing import TYPE_CHECKING, List
from sqlalchemy import Date, DateTime, Enum as SQLEnum, ForeignKey, Integer, Numeric, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from config.database import Base
if TYPE_CHECKING: # pragma: no cover
from .financial_input import FinancialInput
from .project import Project
from .simulation_parameter import SimulationParameter
class ScenarioStatus(str, Enum):
"""Lifecycle states for project scenarios."""
DRAFT = "draft"
ACTIVE = "active"
ARCHIVED = "archived"
class Scenario(Base):
"""A specific configuration of assumptions for a project."""
__tablename__ = "scenarios"
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
project_id: Mapped[int] = mapped_column(
ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
status: Mapped[ScenarioStatus] = mapped_column(
SQLEnum(ScenarioStatus), nullable=False, default=ScenarioStatus.DRAFT
)
start_date: Mapped[date | None] = mapped_column(Date, nullable=True)
end_date: Mapped[date | None] = mapped_column(Date, nullable=True)
discount_rate: Mapped[float | None] = mapped_column(Numeric(5, 2), nullable=True)
currency: Mapped[str | None] = mapped_column(String(3), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
)
project: Mapped["Project"] = relationship("Project", back_populates="scenarios")
financial_inputs: Mapped[List["FinancialInput"]] = relationship(
"FinancialInput",
back_populates="scenario",
cascade="all, delete-orphan",
passive_deletes=True,
)
simulation_parameters: Mapped[List["SimulationParameter"]] = relationship(
"SimulationParameter",
back_populates="scenario",
cascade="all, delete-orphan",
passive_deletes=True,
)
def __repr__(self) -> str: # pragma: no cover
return f"Scenario(id={self.id!r}, name={self.name!r}, project_id={self.project_id!r})"

View File

@@ -0,0 +1,69 @@
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import TYPE_CHECKING
from sqlalchemy import (
JSON,
DateTime,
Enum as SQLEnum,
ForeignKey,
Integer,
Numeric,
String,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from config.database import Base
if TYPE_CHECKING: # pragma: no cover
from .scenario import Scenario
class DistributionType(str, Enum):
"""Supported stochastic distribution families for simulations."""
NORMAL = "normal"
TRIANGULAR = "triangular"
UNIFORM = "uniform"
LOGNORMAL = "lognormal"
CUSTOM = "custom"
class SimulationParameter(Base):
"""Probability distribution settings for scenario simulations."""
__tablename__ = "simulation_parameters"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
scenario_id: Mapped[int] = mapped_column(
ForeignKey("scenarios.id", ondelete="CASCADE"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String(255), nullable=False)
distribution: Mapped[DistributionType] = mapped_column(
SQLEnum(DistributionType), nullable=False
)
mean_value: Mapped[float | None] = mapped_column(Numeric(18, 4), nullable=True)
standard_deviation: Mapped[float | None] = mapped_column(Numeric(18, 4), nullable=True)
minimum_value: Mapped[float | None] = mapped_column(Numeric(18, 4), nullable=True)
maximum_value: Mapped[float | None] = mapped_column(Numeric(18, 4), nullable=True)
unit: Mapped[str | None] = mapped_column(String(32), nullable=True)
metadata: Mapped[dict | None] = mapped_column(JSON, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
)
scenario: Mapped["Scenario"] = relationship(
"Scenario", back_populates="simulation_parameters"
)
def __repr__(self) -> str: # pragma: no cover
return (
f"SimulationParameter(id={self.id!r}, scenario_id={self.scenario_id!r}, "
f"name={self.name!r})"
)