1469 lines
53 KiB
Python
1469 lines
53 KiB
Python
"""Idempotent DB initialization and seeding using Pydantic validation and raw SQL.
|
|
|
|
Usage:
|
|
from scripts.init_db import init_db
|
|
init_db()
|
|
|
|
This module creates PostgreSQL ENUM types if missing, creates minimal tables
|
|
required for bootstrapping (roles, users, user_roles, pricing_settings and
|
|
ancillary pricing tables), and seeds initial rows using INSERT ... ON CONFLICT
|
|
DO NOTHING so it's safe to run multiple times.
|
|
|
|
Notes:
|
|
- This module avoids importing application models at import time to prevent
|
|
side-effects. Database connections are created inside functions.
|
|
- It intentionally performs non-destructive operations only (CREATE IF NOT
|
|
EXISTS, INSERT ... ON CONFLICT).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import List, Optional, Set
|
|
import os
|
|
import logging
|
|
from decimal import Decimal
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from sqlalchemy import JSON, create_engine, text
|
|
from sqlalchemy.engine import Engine
|
|
from passlib.context import CryptContext
|
|
from sqlalchemy.sql import bindparam
|
|
|
|
logger = logging.getLogger(__name__)
|
|
password_context = CryptContext(schemes=["argon2"], deprecated="auto")
|
|
|
|
# ENUM definitions matching previous schema
|
|
ENUM_DEFINITIONS = {
|
|
"miningoperationtype": [
|
|
"open_pit",
|
|
"underground",
|
|
"in_situ_leach",
|
|
"placer",
|
|
"quarry",
|
|
"mountaintop_removal",
|
|
"other",
|
|
],
|
|
"scenariostatus": ["draft", "active", "archived"],
|
|
"financialcategory": ["capex", "opex", "revenue", "contingency", "other"],
|
|
"costbucket": [
|
|
"capital_initial",
|
|
"capital_sustaining",
|
|
"operating_fixed",
|
|
"operating_variable",
|
|
"maintenance",
|
|
"reclamation",
|
|
"royalties",
|
|
"general_admin",
|
|
],
|
|
"distributiontype": ["normal", "triangular", "uniform", "lognormal", "custom"],
|
|
"stochasticvariable": [
|
|
"ore_grade",
|
|
"recovery_rate",
|
|
"metal_price",
|
|
"operating_cost",
|
|
"capital_cost",
|
|
"discount_rate",
|
|
"throughput",
|
|
],
|
|
"resourcetype": [
|
|
"diesel",
|
|
"electricity",
|
|
"water",
|
|
"explosives",
|
|
"reagents",
|
|
"labor",
|
|
"equipment_hours",
|
|
"tailings_capacity",
|
|
],
|
|
}
|
|
|
|
# Minimal DDL for tables we seed / that bootstrap relies on
|
|
|
|
|
|
def _get_table_ddls(is_sqlite: bool) -> List[str]:
|
|
if is_sqlite:
|
|
return [
|
|
# roles
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS roles (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
display_name TEXT NOT NULL,
|
|
description TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
""",
|
|
# users
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
email TEXT NOT NULL UNIQUE,
|
|
username TEXT NOT NULL UNIQUE,
|
|
password_hash TEXT NOT NULL,
|
|
is_active INTEGER NOT NULL DEFAULT 1,
|
|
is_superuser INTEGER NOT NULL DEFAULT 0,
|
|
last_login_at DATETIME,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
""",
|
|
# user_roles
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS user_roles (
|
|
user_id INTEGER NOT NULL,
|
|
role_id INTEGER NOT NULL,
|
|
granted_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
granted_by INTEGER,
|
|
PRIMARY KEY (user_id, role_id)
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS navigation_groups (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
slug TEXT NOT NULL UNIQUE,
|
|
label TEXT NOT NULL,
|
|
sort_order INTEGER NOT NULL DEFAULT 100,
|
|
icon TEXT,
|
|
tooltip TEXT,
|
|
is_enabled INTEGER NOT NULL DEFAULT 1,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS navigation_links (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
group_id INTEGER NOT NULL REFERENCES navigation_groups(id) ON DELETE CASCADE,
|
|
parent_link_id INTEGER REFERENCES navigation_links(id) ON DELETE CASCADE,
|
|
slug TEXT NOT NULL,
|
|
label TEXT NOT NULL,
|
|
route_name TEXT,
|
|
href_override TEXT,
|
|
match_prefix TEXT,
|
|
sort_order INTEGER NOT NULL DEFAULT 100,
|
|
icon TEXT,
|
|
tooltip TEXT,
|
|
required_roles TEXT NOT NULL DEFAULT '[]',
|
|
is_enabled INTEGER NOT NULL DEFAULT 1,
|
|
is_external INTEGER NOT NULL DEFAULT 0,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE (group_id, slug)
|
|
);
|
|
""",
|
|
# pricing_settings
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS pricing_settings (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
slug TEXT NOT NULL UNIQUE,
|
|
description TEXT,
|
|
default_currency TEXT,
|
|
default_payable_pct REAL DEFAULT 100.00 NOT NULL,
|
|
moisture_threshold_pct REAL DEFAULT 8.00 NOT NULL,
|
|
moisture_penalty_per_pct REAL DEFAULT 0.0000 NOT NULL,
|
|
metadata TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
""",
|
|
# pricing_metal_settings
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS pricing_metal_settings (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
pricing_settings_id INTEGER NOT NULL REFERENCES pricing_settings(id) ON DELETE CASCADE,
|
|
metal_code TEXT NOT NULL,
|
|
payable_pct REAL,
|
|
moisture_threshold_pct REAL,
|
|
moisture_penalty_per_pct REAL,
|
|
data TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE (pricing_settings_id, metal_code)
|
|
);
|
|
""",
|
|
# pricing_impurity_settings
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS pricing_impurity_settings (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
pricing_settings_id INTEGER NOT NULL REFERENCES pricing_settings(id) ON DELETE CASCADE,
|
|
impurity_code TEXT NOT NULL,
|
|
threshold_ppm REAL DEFAULT 0.0000 NOT NULL,
|
|
penalty_per_ppm REAL DEFAULT 0.0000 NOT NULL,
|
|
notes TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE (pricing_settings_id, impurity_code)
|
|
);
|
|
""",
|
|
# core domain tables: projects, scenarios, financial_inputs, simulation_parameters
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS projects (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
location TEXT,
|
|
operation_type TEXT NOT NULL CHECK (operation_type IN ('open_pit', 'underground', 'in_situ_leach', 'placer', 'quarry', 'mountaintop_removal', 'other')),
|
|
description TEXT,
|
|
pricing_settings_id INTEGER REFERENCES pricing_settings(id) ON DELETE SET NULL,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS scenarios (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
|
|
name TEXT NOT NULL,
|
|
description TEXT,
|
|
status TEXT NOT NULL CHECK (status IN ('draft', 'active', 'archived')),
|
|
start_date DATE,
|
|
end_date DATE,
|
|
discount_rate REAL,
|
|
currency TEXT,
|
|
primary_resource TEXT CHECK (primary_resource IN ('diesel', 'electricity', 'water', 'explosives', 'reagents', 'labor', 'equipment_hours', 'tailings_capacity') OR primary_resource IS NULL),
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE (project_id, name)
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS financial_inputs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
scenario_id INTEGER NOT NULL REFERENCES scenarios(id) ON DELETE CASCADE,
|
|
name TEXT NOT NULL,
|
|
category TEXT NOT NULL CHECK (category IN ('capex', 'opex', 'revenue', 'contingency', 'other')),
|
|
cost_bucket TEXT CHECK (cost_bucket IN ('capital_initial', 'capital_sustaining', 'operating_fixed', 'operating_variable', 'maintenance', 'reclamation', 'royalties', 'general_admin') OR cost_bucket IS NULL),
|
|
amount REAL NOT NULL,
|
|
currency TEXT,
|
|
effective_date DATE,
|
|
notes TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE (scenario_id, name)
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS simulation_parameters (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
scenario_id INTEGER NOT NULL REFERENCES scenarios(id) ON DELETE CASCADE,
|
|
name TEXT NOT NULL,
|
|
distribution TEXT NOT NULL CHECK (distribution IN ('normal', 'triangular', 'uniform', 'lognormal', 'custom')),
|
|
variable TEXT CHECK (variable IN ('ore_grade', 'recovery_rate', 'metal_price', 'operating_cost', 'capital_cost', 'discount_rate', 'throughput') OR variable IS NULL),
|
|
resource_type TEXT CHECK (resource_type IN ('diesel', 'electricity', 'water', 'explosives', 'reagents', 'labor', 'equipment_hours', 'tailings_capacity') OR resource_type IS NULL),
|
|
mean_value REAL,
|
|
standard_deviation REAL,
|
|
minimum_value REAL,
|
|
maximum_value REAL,
|
|
unit TEXT,
|
|
configuration TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
""",
|
|
]
|
|
else:
|
|
# PostgreSQL DDLs
|
|
return [
|
|
# roles
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS roles (
|
|
id INTEGER PRIMARY KEY,
|
|
name VARCHAR(64) NOT NULL,
|
|
display_name VARCHAR(128) NOT NULL,
|
|
description TEXT,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_roles_name UNIQUE (name)
|
|
);
|
|
""",
|
|
# users
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id SERIAL PRIMARY KEY,
|
|
email VARCHAR(255) NOT NULL,
|
|
username VARCHAR(128) NOT NULL,
|
|
password_hash VARCHAR(255) NOT NULL,
|
|
is_active BOOLEAN NOT NULL DEFAULT true,
|
|
is_superuser BOOLEAN NOT NULL DEFAULT false,
|
|
last_login_at TIMESTAMPTZ,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_users_email UNIQUE (email),
|
|
CONSTRAINT uq_users_username UNIQUE (username)
|
|
);
|
|
""",
|
|
# user_roles
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS user_roles (
|
|
user_id INTEGER NOT NULL,
|
|
role_id INTEGER NOT NULL,
|
|
granted_at TIMESTAMPTZ DEFAULT now(),
|
|
granted_by INTEGER,
|
|
PRIMARY KEY (user_id, role_id),
|
|
CONSTRAINT uq_user_roles_user_role UNIQUE (user_id, role_id)
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS navigation_groups (
|
|
id SERIAL PRIMARY KEY,
|
|
slug VARCHAR(64) NOT NULL,
|
|
label VARCHAR(128) NOT NULL,
|
|
sort_order INTEGER NOT NULL DEFAULT 100,
|
|
icon VARCHAR(64),
|
|
tooltip VARCHAR(255),
|
|
is_enabled BOOLEAN NOT NULL DEFAULT true,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_navigation_groups_slug UNIQUE (slug)
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS navigation_links (
|
|
id SERIAL PRIMARY KEY,
|
|
group_id INTEGER NOT NULL REFERENCES navigation_groups(id) ON DELETE CASCADE,
|
|
parent_link_id INTEGER REFERENCES navigation_links(id) ON DELETE CASCADE,
|
|
slug VARCHAR(64) NOT NULL,
|
|
label VARCHAR(128) NOT NULL,
|
|
route_name VARCHAR(128),
|
|
href_override VARCHAR(512),
|
|
match_prefix VARCHAR(512),
|
|
sort_order INTEGER NOT NULL DEFAULT 100,
|
|
icon VARCHAR(64),
|
|
tooltip VARCHAR(255),
|
|
required_roles JSONB NOT NULL DEFAULT '[]'::jsonb,
|
|
is_enabled BOOLEAN NOT NULL DEFAULT true,
|
|
is_external BOOLEAN NOT NULL DEFAULT false,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_navigation_links_group_slug UNIQUE (group_id, slug)
|
|
);
|
|
""",
|
|
# pricing_settings
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS pricing_settings (
|
|
id SERIAL PRIMARY KEY,
|
|
name VARCHAR(128) NOT NULL,
|
|
slug VARCHAR(64) NOT NULL,
|
|
description TEXT,
|
|
default_currency VARCHAR(3),
|
|
default_payable_pct NUMERIC(5,2) DEFAULT 100.00 NOT NULL,
|
|
moisture_threshold_pct NUMERIC(5,2) DEFAULT 8.00 NOT NULL,
|
|
moisture_penalty_per_pct NUMERIC(14,4) DEFAULT 0.0000 NOT NULL,
|
|
metadata JSONB,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_pricing_settings_slug UNIQUE (slug),
|
|
CONSTRAINT uq_pricing_settings_name UNIQUE (name)
|
|
);
|
|
""",
|
|
# pricing_metal_settings
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS pricing_metal_settings (
|
|
id SERIAL PRIMARY KEY,
|
|
pricing_settings_id INTEGER NOT NULL REFERENCES pricing_settings(id) ON DELETE CASCADE,
|
|
metal_code VARCHAR(32) NOT NULL,
|
|
payable_pct NUMERIC(5,2),
|
|
moisture_threshold_pct NUMERIC(5,2),
|
|
moisture_penalty_per_pct NUMERIC(14,4),
|
|
data JSONB,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_pricing_metal_settings_code UNIQUE (pricing_settings_id, metal_code)
|
|
);
|
|
""",
|
|
# pricing_impurity_settings
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS pricing_impurity_settings (
|
|
id SERIAL PRIMARY KEY,
|
|
pricing_settings_id INTEGER NOT NULL REFERENCES pricing_settings(id) ON DELETE CASCADE,
|
|
impurity_code VARCHAR(32) NOT NULL,
|
|
threshold_ppm NUMERIC(14,4) DEFAULT 0.0000 NOT NULL,
|
|
penalty_per_ppm NUMERIC(14,4) DEFAULT 0.0000 NOT NULL,
|
|
notes TEXT,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_pricing_impurity_settings_code UNIQUE (pricing_settings_id, impurity_code)
|
|
);
|
|
""",
|
|
# core domain tables: projects, scenarios, financial_inputs, simulation_parameters
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS projects (
|
|
id SERIAL PRIMARY KEY,
|
|
name VARCHAR(255) NOT NULL,
|
|
location VARCHAR(255),
|
|
operation_type miningoperationtype NOT NULL,
|
|
description TEXT,
|
|
pricing_settings_id INTEGER REFERENCES pricing_settings(id) ON DELETE SET NULL,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_projects_name UNIQUE (name)
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS scenarios (
|
|
id SERIAL PRIMARY KEY,
|
|
project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
|
|
name VARCHAR(255) NOT NULL,
|
|
description TEXT,
|
|
status scenariostatus NOT NULL,
|
|
start_date DATE,
|
|
end_date DATE,
|
|
discount_rate NUMERIC(5,2),
|
|
currency VARCHAR(3),
|
|
primary_resource resourcetype,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_scenarios_project_name UNIQUE (project_id, name)
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS financial_inputs (
|
|
id SERIAL PRIMARY KEY,
|
|
scenario_id INTEGER NOT NULL REFERENCES scenarios(id) ON DELETE CASCADE,
|
|
name VARCHAR(255) NOT NULL,
|
|
category financialcategory NOT NULL,
|
|
cost_bucket costbucket,
|
|
amount NUMERIC(18,2) NOT NULL,
|
|
currency VARCHAR(3),
|
|
effective_date DATE,
|
|
notes TEXT,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now(),
|
|
CONSTRAINT uq_financial_inputs_scenario_name UNIQUE (scenario_id, name)
|
|
);
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS simulation_parameters (
|
|
id SERIAL PRIMARY KEY,
|
|
scenario_id INTEGER NOT NULL REFERENCES scenarios(id) ON DELETE CASCADE,
|
|
name VARCHAR(255) NOT NULL,
|
|
distribution distributiontype NOT NULL,
|
|
variable stochasticvariable,
|
|
resource_type resourcetype,
|
|
mean_value NUMERIC(18,4),
|
|
standard_deviation NUMERIC(18,4),
|
|
minimum_value NUMERIC(18,4),
|
|
maximum_value NUMERIC(18,4),
|
|
unit VARCHAR(32),
|
|
configuration JSONB,
|
|
created_at TIMESTAMPTZ DEFAULT now(),
|
|
updated_at TIMESTAMPTZ DEFAULT now()
|
|
);
|
|
""",
|
|
]
|
|
|
|
|
|
# Seeds
|
|
TABLE_DDLS: List[str] = _get_table_ddls(is_sqlite=False)
|
|
|
|
|
|
DEFAULT_ROLES = [
|
|
{"id": 1, "name": "admin", "display_name": "Administrator",
|
|
"description": "Full platform access with user management rights."},
|
|
{"id": 2, "name": "project_manager", "display_name": "Project Manager",
|
|
"description": "Manage projects, scenarios, and associated data."},
|
|
{"id": 3, "name": "analyst", "display_name": "Analyst",
|
|
"description": "Review dashboards and scenario outputs."},
|
|
{"id": 4, "name": "viewer", "display_name": "Viewer",
|
|
"description": "Read-only access to assigned projects and reports."},
|
|
]
|
|
|
|
DEFAULT_ADMIN = {"id": 1, "email": "admin@calminer.local", "username": "admin",
|
|
"password": "ChangeMe123!", "is_active": True, "is_superuser": True}
|
|
DEFAULT_PRICING = {
|
|
"slug": "default",
|
|
"name": "Default Pricing",
|
|
"description": "Automatically generated default pricing settings.",
|
|
"default_currency": "USD",
|
|
"default_payable_pct": 100.0,
|
|
"moisture_threshold_pct": 8.0,
|
|
"moisture_penalty_per_pct": 0.0,
|
|
}
|
|
|
|
|
|
class ProjectSeed(BaseModel):
|
|
name: str
|
|
location: str | None = None
|
|
operation_type: str
|
|
description: str | None = None
|
|
|
|
|
|
class ScenarioSeed(BaseModel):
|
|
project_name: str
|
|
name: str
|
|
description: str | None = None
|
|
status: str = "active"
|
|
discount_rate: float | None = Field(default=None)
|
|
currency: str | None = Field(default="USD")
|
|
primary_resource: str | None = Field(default=None)
|
|
|
|
|
|
class FinancialInputSeed(BaseModel):
|
|
scenario_name: str
|
|
project_name: str
|
|
name: str
|
|
category: str
|
|
cost_bucket: str | None = None
|
|
amount: Decimal
|
|
currency: str = "USD"
|
|
notes: str | None = None
|
|
|
|
|
|
class RoleSeed(BaseModel):
|
|
id: int
|
|
name: str
|
|
display_name: str
|
|
description: Optional[str]
|
|
|
|
|
|
class UserSeed(BaseModel):
|
|
id: int
|
|
email: str
|
|
username: str
|
|
password: str
|
|
is_active: bool = True
|
|
is_superuser: bool = False
|
|
|
|
@field_validator("password")
|
|
def password_min_len(cls, v: str) -> str:
|
|
if not v or len(v) < 8:
|
|
raise ValueError("password must be at least 8 characters")
|
|
return v
|
|
|
|
|
|
class PricingSeed(BaseModel):
|
|
slug: str
|
|
name: str
|
|
description: Optional[str]
|
|
default_currency: Optional[str]
|
|
default_payable_pct: float
|
|
moisture_threshold_pct: float
|
|
moisture_penalty_per_pct: float
|
|
|
|
|
|
class NavigationGroupSeed(BaseModel):
|
|
slug: str
|
|
label: str
|
|
sort_order: int = 100
|
|
icon: Optional[str] = None
|
|
tooltip: Optional[str] = None
|
|
is_enabled: bool = True
|
|
|
|
|
|
class NavigationLinkSeed(BaseModel):
|
|
slug: str
|
|
group_slug: str
|
|
label: str
|
|
route_name: Optional[str] = None
|
|
href_override: Optional[str] = None
|
|
match_prefix: Optional[str] = None
|
|
sort_order: int = 100
|
|
icon: Optional[str] = None
|
|
tooltip: Optional[str] = None
|
|
required_roles: list[str] = Field(default_factory=list)
|
|
is_enabled: bool = True
|
|
is_external: bool = False
|
|
parent_slug: Optional[str] = None
|
|
|
|
@field_validator("required_roles", mode="after")
|
|
def _normalise_roles(cls, value: list[str]) -> list[str]:
|
|
normalised = []
|
|
for role in value:
|
|
if not role:
|
|
continue
|
|
slug = role.strip().lower()
|
|
if slug and slug not in normalised:
|
|
normalised.append(slug)
|
|
return normalised
|
|
|
|
@field_validator("route_name")
|
|
def _route_or_href(cls, value: Optional[str], info):
|
|
href = info.data.get("href_override")
|
|
if not value and not href:
|
|
raise ValueError(
|
|
"navigation link requires route_name or href_override")
|
|
return value
|
|
|
|
|
|
DEFAULT_NAVIGATION_GROUPS: list[NavigationGroupSeed] = [
|
|
NavigationGroupSeed(
|
|
slug="workspace",
|
|
label="Workspace",
|
|
sort_order=10,
|
|
icon="briefcase",
|
|
tooltip="Primary work hub",
|
|
),
|
|
NavigationGroupSeed(
|
|
slug="insights",
|
|
label="Insights",
|
|
sort_order=20,
|
|
icon="insights",
|
|
tooltip="Analytics and reports",
|
|
),
|
|
NavigationGroupSeed(
|
|
slug="configuration",
|
|
label="Configuration",
|
|
sort_order=30,
|
|
icon="cog",
|
|
tooltip="Administration and settings",
|
|
),
|
|
NavigationGroupSeed(
|
|
slug="account",
|
|
label="Account",
|
|
sort_order=40,
|
|
icon="user",
|
|
tooltip="Session management",
|
|
),
|
|
]
|
|
|
|
|
|
DEFAULT_NAVIGATION_LINKS: list[NavigationLinkSeed] = [
|
|
NavigationLinkSeed(
|
|
slug="dashboard",
|
|
group_slug="workspace",
|
|
label="Dashboard",
|
|
route_name="dashboard.home",
|
|
match_prefix="/",
|
|
sort_order=10,
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="projects",
|
|
group_slug="workspace",
|
|
label="Projects",
|
|
route_name="projects.project_list_page",
|
|
match_prefix="/projects",
|
|
sort_order=20,
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="project-create",
|
|
group_slug="workspace",
|
|
label="New Project",
|
|
route_name="projects.create_project_form",
|
|
match_prefix="/projects/create",
|
|
sort_order=30,
|
|
required_roles=["project_manager", "admin"],
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="imports",
|
|
group_slug="workspace",
|
|
label="Imports",
|
|
href_override="/imports/ui",
|
|
match_prefix="/imports",
|
|
sort_order=40,
|
|
required_roles=["analyst", "admin"],
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="profitability",
|
|
group_slug="workspace",
|
|
label="Profitability Calculator",
|
|
route_name="calculations.profitability_form",
|
|
href_override="/calculations/profitability",
|
|
match_prefix="/calculations/profitability",
|
|
sort_order=50,
|
|
required_roles=["analyst", "admin"],
|
|
parent_slug="projects",
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="opex",
|
|
group_slug="workspace",
|
|
label="Opex Planner",
|
|
route_name="calculations.opex_form",
|
|
href_override="/calculations/opex",
|
|
match_prefix="/calculations/opex",
|
|
sort_order=60,
|
|
required_roles=["analyst", "admin"],
|
|
parent_slug="projects",
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="capex",
|
|
group_slug="workspace",
|
|
label="Capex Planner",
|
|
route_name="calculations.capex_form",
|
|
href_override="/calculations/capex",
|
|
match_prefix="/calculations/capex",
|
|
sort_order=70,
|
|
required_roles=["analyst", "admin"],
|
|
parent_slug="projects",
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="simulations",
|
|
group_slug="insights",
|
|
label="Simulations",
|
|
href_override="/ui/simulations",
|
|
match_prefix="/ui/simulations",
|
|
sort_order=10,
|
|
required_roles=["analyst", "admin"],
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="reporting",
|
|
group_slug="insights",
|
|
label="Reporting",
|
|
href_override="/ui/reporting",
|
|
match_prefix="/ui/reporting",
|
|
sort_order=20,
|
|
required_roles=["analyst", "admin"],
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="settings",
|
|
group_slug="configuration",
|
|
label="Settings",
|
|
href_override="/ui/settings",
|
|
match_prefix="/ui/settings",
|
|
sort_order=10,
|
|
required_roles=["admin"],
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="themes",
|
|
group_slug="configuration",
|
|
label="Themes",
|
|
href_override="/theme-settings",
|
|
match_prefix="/theme-settings",
|
|
sort_order=20,
|
|
required_roles=["admin"],
|
|
parent_slug="settings",
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="currencies",
|
|
group_slug="configuration",
|
|
label="Currency Management",
|
|
href_override="/ui/currencies",
|
|
match_prefix="/ui/currencies",
|
|
sort_order=30,
|
|
required_roles=["admin"],
|
|
parent_slug="settings",
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="logout",
|
|
group_slug="account",
|
|
label="Logout",
|
|
route_name="auth.logout",
|
|
match_prefix="/logout",
|
|
sort_order=10,
|
|
required_roles=["viewer", "analyst", "project_manager", "admin"],
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="login",
|
|
group_slug="account",
|
|
label="Login",
|
|
route_name="auth.login_form",
|
|
match_prefix="/login",
|
|
sort_order=10,
|
|
required_roles=["anonymous"],
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="register",
|
|
group_slug="account",
|
|
label="Register",
|
|
route_name="auth.register_form",
|
|
match_prefix="/register",
|
|
sort_order=20,
|
|
required_roles=["anonymous"],
|
|
),
|
|
NavigationLinkSeed(
|
|
slug="forgot-password",
|
|
group_slug="account",
|
|
label="Forgot Password",
|
|
route_name="auth.password_reset_request_form",
|
|
match_prefix="/forgot-password",
|
|
sort_order=30,
|
|
required_roles=["anonymous"],
|
|
),
|
|
]
|
|
|
|
|
|
DEFAULT_PROJECTS: list[ProjectSeed] = [
|
|
ProjectSeed(
|
|
name="Helios Copper",
|
|
location="Chile",
|
|
operation_type="open_pit",
|
|
description="Flagship open pit copper operation used for demos",
|
|
),
|
|
ProjectSeed(
|
|
name="Luna Nickel",
|
|
location="Australia",
|
|
operation_type="underground",
|
|
description="Underground nickel sulphide project with stochastic modelling",
|
|
),
|
|
]
|
|
|
|
|
|
DEFAULT_SCENARIOS: list[ScenarioSeed] = [
|
|
ScenarioSeed(
|
|
project_name="Helios Copper",
|
|
name="Base Case",
|
|
description="Deterministic base case for Helios",
|
|
status="active",
|
|
discount_rate=8.0,
|
|
primary_resource="diesel",
|
|
),
|
|
ScenarioSeed(
|
|
project_name="Helios Copper",
|
|
name="Expansion Case",
|
|
description="Expansion scenario with increased throughput",
|
|
status="draft",
|
|
discount_rate=9.0,
|
|
primary_resource="electricity",
|
|
),
|
|
ScenarioSeed(
|
|
project_name="Luna Nickel",
|
|
name="Feasibility",
|
|
description="Feasibility scenario targeting steady state",
|
|
status="active",
|
|
discount_rate=10.0,
|
|
primary_resource="electricity",
|
|
),
|
|
]
|
|
|
|
|
|
DEFAULT_FINANCIAL_INPUTS: list[FinancialInputSeed] = [
|
|
FinancialInputSeed(
|
|
project_name="Helios Copper",
|
|
scenario_name="Base Case",
|
|
name="Initial Capital",
|
|
category="capex",
|
|
cost_bucket="capital_initial",
|
|
amount=Decimal("450000000"),
|
|
notes="Initial mine development costs",
|
|
),
|
|
FinancialInputSeed(
|
|
project_name="Helios Copper",
|
|
scenario_name="Base Case",
|
|
name="Opex",
|
|
category="opex",
|
|
cost_bucket="operating_variable",
|
|
amount=Decimal("75000000"),
|
|
notes="Annual processing operating expenditure",
|
|
),
|
|
FinancialInputSeed(
|
|
project_name="Helios Copper",
|
|
scenario_name="Expansion Case",
|
|
name="Expansion Capital",
|
|
category="capex",
|
|
cost_bucket="capital_sustaining",
|
|
amount=Decimal("120000000"),
|
|
),
|
|
FinancialInputSeed(
|
|
project_name="Luna Nickel",
|
|
scenario_name="Feasibility",
|
|
name="Nickel Revenue",
|
|
category="revenue",
|
|
cost_bucket=None,
|
|
amount=Decimal("315000000"),
|
|
),
|
|
]
|
|
|
|
|
|
def _get_database_url() -> str:
|
|
# Prefer the same DATABASE_URL used by the application
|
|
from config.database import DATABASE_URL
|
|
|
|
return DATABASE_URL
|
|
|
|
|
|
def _is_sqlite(database_url: str) -> bool:
|
|
return database_url.startswith("sqlite://")
|
|
|
|
|
|
def _create_engine(database_url: Optional[str] = None) -> Engine:
|
|
database_url = database_url or _get_database_url()
|
|
engine = create_engine(database_url, future=True)
|
|
return engine
|
|
|
|
|
|
def _create_enum_if_missing_sql(type_name: str, values: List[str]) -> str:
|
|
# Use a DO block to safely create the enum only if it is missing
|
|
vals = ", ".join(f"'{v}'" for v in values)
|
|
sql = (
|
|
"DO $$ BEGIN "
|
|
f"IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = '{type_name}') THEN "
|
|
f"CREATE TYPE {type_name} AS ENUM ({vals}); "
|
|
"END IF; END $$;"
|
|
)
|
|
return sql
|
|
|
|
|
|
def ensure_enums(engine: Engine, is_sqlite: bool) -> None:
|
|
if is_sqlite:
|
|
# SQLite doesn't have enums, constraints are in table DDL
|
|
logger.debug("Skipping enum creation for SQLite")
|
|
return
|
|
with engine.begin() as conn:
|
|
for name, vals in ENUM_DEFINITIONS.items():
|
|
sql = _create_enum_if_missing_sql(name, vals)
|
|
logger.debug("Ensuring enum %s: %s", name, sql)
|
|
conn.execute(text(sql))
|
|
|
|
|
|
def _fetch_enum_values(conn, type_name: str) -> Set[str]:
|
|
rows = conn.execute(
|
|
text(
|
|
"""
|
|
SELECT e.enumlabel
|
|
FROM pg_enum e
|
|
JOIN pg_type t ON t.oid = e.enumtypid
|
|
WHERE t.typname = :type_name
|
|
"""
|
|
),
|
|
{"type_name": type_name},
|
|
)
|
|
return {row.enumlabel for row in rows}
|
|
|
|
|
|
def normalize_enum_values(engine: Engine, is_sqlite: bool) -> None:
|
|
if is_sqlite:
|
|
# No enums to normalize in SQLite
|
|
logger.debug("Skipping enum normalization for SQLite")
|
|
return
|
|
with engine.begin() as conn:
|
|
for type_name, expected_values in ENUM_DEFINITIONS.items():
|
|
try:
|
|
existing_values = _fetch_enum_values(conn, type_name)
|
|
except Exception as exc: # pragma: no cover - system catalogs missing
|
|
logger.debug(
|
|
"Skipping enum normalization for %s due to error: %s",
|
|
type_name,
|
|
exc,
|
|
)
|
|
continue
|
|
|
|
expected_set = set(expected_values)
|
|
for value in list(existing_values):
|
|
if value in expected_set:
|
|
continue
|
|
|
|
normalized = value.lower()
|
|
if (
|
|
normalized != value
|
|
and normalized in expected_set
|
|
and normalized not in existing_values
|
|
):
|
|
logger.info(
|
|
"Renaming enum value %s.%s -> %s",
|
|
type_name,
|
|
value,
|
|
normalized,
|
|
)
|
|
conn.execute(
|
|
text(
|
|
f"ALTER TYPE {type_name} RENAME VALUE :old_value TO :new_value"
|
|
),
|
|
{"old_value": value, "new_value": normalized},
|
|
)
|
|
existing_values.remove(value)
|
|
existing_values.add(normalized)
|
|
|
|
|
|
def ensure_tables(engine: Engine, is_sqlite: bool) -> None:
|
|
table_ddls = _get_table_ddls(is_sqlite)
|
|
with engine.begin() as conn:
|
|
for ddl in table_ddls:
|
|
logger.debug("Executing DDL:\n%s", ddl)
|
|
conn.execute(text(ddl))
|
|
|
|
|
|
CONSTRAINT_DDLS = [
|
|
"""
|
|
DO $$
|
|
BEGIN
|
|
IF NOT EXISTS (
|
|
SELECT 1
|
|
FROM pg_constraint
|
|
WHERE conname = 'uq_scenarios_project_name'
|
|
) THEN
|
|
ALTER TABLE scenarios
|
|
ADD CONSTRAINT uq_scenarios_project_name UNIQUE (project_id, name);
|
|
END IF;
|
|
END;
|
|
$$;
|
|
""",
|
|
"""
|
|
DO $$
|
|
BEGIN
|
|
IF NOT EXISTS (
|
|
SELECT 1
|
|
FROM pg_constraint
|
|
WHERE conname = 'uq_financial_inputs_scenario_name'
|
|
) THEN
|
|
ALTER TABLE financial_inputs
|
|
ADD CONSTRAINT uq_financial_inputs_scenario_name UNIQUE (scenario_id, name);
|
|
END IF;
|
|
END;
|
|
$$;
|
|
""",
|
|
]
|
|
|
|
|
|
def ensure_constraints(engine: Engine, is_sqlite: bool) -> None:
|
|
if is_sqlite:
|
|
# Constraints are already in table DDL for SQLite
|
|
logger.debug("Skipping constraint creation for SQLite")
|
|
return
|
|
with engine.begin() as conn:
|
|
for ddl in CONSTRAINT_DDLS:
|
|
logger.debug("Ensuring constraint via:\n%s", ddl)
|
|
conn.execute(text(ddl))
|
|
|
|
|
|
def seed_roles(engine: Engine, is_sqlite: bool) -> None:
|
|
with engine.begin() as conn:
|
|
for r in DEFAULT_ROLES:
|
|
seed = RoleSeed(**r)
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO roles (id, name, display_name, description) VALUES (:id, :name, :display_name, :description) "
|
|
"ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, display_name = EXCLUDED.display_name, description = EXCLUDED.description"
|
|
),
|
|
dict(id=seed.id, name=seed.name,
|
|
display_name=seed.display_name, description=seed.description),
|
|
)
|
|
|
|
|
|
def seed_admin_user(engine: Engine, is_sqlite: bool) -> None:
|
|
with engine.begin() as conn:
|
|
# Use environment-configured admin settings when present so initializer
|
|
# aligns with the application's bootstrap configuration.
|
|
admin_email = os.getenv(
|
|
"CALMINER_SEED_ADMIN_EMAIL", DEFAULT_ADMIN["email"])
|
|
admin_username = os.getenv(
|
|
"CALMINER_SEED_ADMIN_USERNAME", DEFAULT_ADMIN["username"])
|
|
admin_password = os.getenv(
|
|
"CALMINER_SEED_ADMIN_PASSWORD", DEFAULT_ADMIN["password"])
|
|
u = UserSeed(
|
|
id=DEFAULT_ADMIN.get("id", 1),
|
|
email=admin_email,
|
|
username=admin_username,
|
|
password=admin_password,
|
|
is_active=DEFAULT_ADMIN.get("is_active", True),
|
|
is_superuser=DEFAULT_ADMIN.get("is_superuser", True),
|
|
)
|
|
password_hash = password_context.hash(u.password)
|
|
# Upsert by username to avoid conflicting with different admin email configs
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO users (email, username, password_hash, is_active, is_superuser) "
|
|
"VALUES (:email, :username, :password_hash, :is_active, :is_superuser) "
|
|
"ON CONFLICT (username) DO UPDATE SET email = EXCLUDED.email, password_hash = EXCLUDED.password_hash, is_active = EXCLUDED.is_active, is_superuser = EXCLUDED.is_superuser"
|
|
),
|
|
dict(email=u.email, username=u.username, password_hash=password_hash,
|
|
is_active=u.is_active, is_superuser=u.is_superuser),
|
|
)
|
|
# ensure admin has admin role
|
|
# Resolve user_id for role assignment: select by username
|
|
row = conn.execute(text("SELECT id FROM users WHERE username = :username"), dict(
|
|
username=u.username)).fetchone()
|
|
if row is not None:
|
|
user_id = row.id
|
|
else:
|
|
user_id = None
|
|
if user_id is not None:
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO user_roles (user_id, role_id, granted_by) VALUES (:user_id, :role_id, :granted_by) "
|
|
"ON CONFLICT (user_id, role_id) DO NOTHING"
|
|
),
|
|
dict(user_id=user_id, role_id=1, granted_by=user_id),
|
|
)
|
|
|
|
|
|
def ensure_default_pricing(engine: Engine, is_sqlite: bool) -> None:
|
|
with engine.begin() as conn:
|
|
p = PricingSeed(**DEFAULT_PRICING)
|
|
# Try insert on slug conflict
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO pricing_settings (slug, name, description, default_currency, default_payable_pct, moisture_threshold_pct, moisture_penalty_per_pct) "
|
|
"VALUES (:slug, :name, :description, :default_currency, :default_payable_pct, :moisture_threshold_pct, :moisture_penalty_per_pct) "
|
|
"ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name"
|
|
),
|
|
dict(
|
|
slug=p.slug,
|
|
name=p.name,
|
|
description=p.description,
|
|
default_currency=p.default_currency,
|
|
default_payable_pct=p.default_payable_pct,
|
|
moisture_threshold_pct=p.moisture_threshold_pct,
|
|
moisture_penalty_per_pct=p.moisture_penalty_per_pct,
|
|
),
|
|
)
|
|
|
|
|
|
def seed_navigation(engine: Engine, is_sqlite: bool) -> None:
|
|
group_insert_sql = text(
|
|
"""
|
|
INSERT INTO navigation_groups (slug, label, sort_order, icon, tooltip, is_enabled)
|
|
VALUES (:slug, :label, :sort_order, :icon, :tooltip, :is_enabled)
|
|
ON CONFLICT (slug) DO UPDATE SET
|
|
label = EXCLUDED.label,
|
|
sort_order = EXCLUDED.sort_order,
|
|
icon = EXCLUDED.icon,
|
|
tooltip = EXCLUDED.tooltip,
|
|
is_enabled = EXCLUDED.is_enabled
|
|
"""
|
|
)
|
|
|
|
link_insert_sql = text(
|
|
"""
|
|
INSERT INTO navigation_links (
|
|
group_id, parent_link_id, slug, label, route_name, href_override,
|
|
match_prefix, sort_order, icon, tooltip, required_roles, is_enabled, is_external
|
|
)
|
|
VALUES (
|
|
:group_id, :parent_link_id, :slug, :label, :route_name, :href_override,
|
|
:match_prefix, :sort_order, :icon, :tooltip, :required_roles, :is_enabled, :is_external
|
|
)
|
|
ON CONFLICT (group_id, slug) DO UPDATE SET
|
|
parent_link_id = EXCLUDED.parent_link_id,
|
|
label = EXCLUDED.label,
|
|
route_name = EXCLUDED.route_name,
|
|
href_override = EXCLUDED.href_override,
|
|
match_prefix = EXCLUDED.match_prefix,
|
|
sort_order = EXCLUDED.sort_order,
|
|
icon = EXCLUDED.icon,
|
|
tooltip = EXCLUDED.tooltip,
|
|
required_roles = EXCLUDED.required_roles,
|
|
is_enabled = EXCLUDED.is_enabled,
|
|
is_external = EXCLUDED.is_external
|
|
"""
|
|
)
|
|
link_insert_sql = link_insert_sql.bindparams(
|
|
bindparam("required_roles", type_=JSON)
|
|
)
|
|
|
|
with engine.begin() as conn:
|
|
role_rows = conn.execute(text("SELECT name FROM roles")).fetchall()
|
|
available_roles = {row.name for row in role_rows}
|
|
|
|
def resolve_roles(raw_roles: list[str]) -> list[str]:
|
|
if not raw_roles:
|
|
return []
|
|
|
|
resolved: list[str] = []
|
|
missing: list[str] = []
|
|
for slug in raw_roles:
|
|
if slug == "anonymous":
|
|
if slug not in resolved:
|
|
resolved.append(slug)
|
|
continue
|
|
if slug in available_roles:
|
|
if slug not in resolved:
|
|
resolved.append(slug)
|
|
else:
|
|
missing.append(slug)
|
|
|
|
if missing:
|
|
logger.warning(
|
|
"Navigation seed roles %s are missing; defaulting link access to admin only",
|
|
", ".join(missing),
|
|
)
|
|
if "admin" in available_roles and "admin" not in resolved:
|
|
resolved.append("admin")
|
|
|
|
return resolved
|
|
|
|
group_ids: dict[str, int] = {}
|
|
for group_seed in DEFAULT_NAVIGATION_GROUPS:
|
|
conn.execute(
|
|
group_insert_sql,
|
|
group_seed.model_dump(),
|
|
)
|
|
row = conn.execute(
|
|
text("SELECT id FROM navigation_groups WHERE slug = :slug"),
|
|
{"slug": group_seed.slug},
|
|
).fetchone()
|
|
if row is not None:
|
|
group_ids[group_seed.slug] = row.id
|
|
|
|
if not group_ids:
|
|
logger.warning(
|
|
"Navigation seeding skipped because no groups were inserted")
|
|
return
|
|
|
|
link_ids: dict[tuple[str, str], int] = {}
|
|
parent_pending: list[NavigationLinkSeed] = []
|
|
|
|
for link_seed in DEFAULT_NAVIGATION_LINKS:
|
|
if link_seed.parent_slug:
|
|
parent_pending.append(link_seed)
|
|
continue
|
|
|
|
group_id = group_ids.get(link_seed.group_slug)
|
|
if group_id is None:
|
|
logger.warning(
|
|
"Skipping navigation link '%s' because group '%s' is missing",
|
|
link_seed.slug,
|
|
link_seed.group_slug,
|
|
)
|
|
continue
|
|
|
|
resolved_roles = resolve_roles(link_seed.required_roles)
|
|
|
|
payload = {
|
|
"group_id": group_id,
|
|
"parent_link_id": None,
|
|
"slug": link_seed.slug,
|
|
"label": link_seed.label,
|
|
"route_name": link_seed.route_name,
|
|
"href_override": link_seed.href_override,
|
|
"match_prefix": link_seed.match_prefix,
|
|
"sort_order": link_seed.sort_order,
|
|
"icon": link_seed.icon,
|
|
"tooltip": link_seed.tooltip,
|
|
"required_roles": resolved_roles,
|
|
"is_enabled": link_seed.is_enabled,
|
|
"is_external": link_seed.is_external,
|
|
}
|
|
conn.execute(link_insert_sql, payload)
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT id FROM navigation_links WHERE group_id = :group_id AND slug = :slug"
|
|
),
|
|
{"group_id": group_id, "slug": link_seed.slug},
|
|
).fetchone()
|
|
if row is not None:
|
|
link_ids[(link_seed.group_slug, link_seed.slug)] = row.id
|
|
|
|
for link_seed in parent_pending:
|
|
group_id = group_ids.get(link_seed.group_slug)
|
|
if group_id is None:
|
|
logger.warning(
|
|
"Skipping child navigation link '%s' because group '%s' is missing",
|
|
link_seed.slug,
|
|
link_seed.group_slug,
|
|
)
|
|
continue
|
|
|
|
parent_key = (link_seed.group_slug, link_seed.parent_slug or "")
|
|
parent_id = link_ids.get(parent_key)
|
|
if parent_id is None:
|
|
parent_row = conn.execute(
|
|
text(
|
|
"SELECT id FROM navigation_links WHERE group_id = :group_id AND slug = :slug"
|
|
),
|
|
{"group_id": group_id, "slug": link_seed.parent_slug},
|
|
).fetchone()
|
|
parent_id = parent_row.id if parent_row else None
|
|
|
|
if parent_id is None:
|
|
logger.warning(
|
|
"Skipping child navigation link '%s' because parent '%s' is missing",
|
|
link_seed.slug,
|
|
link_seed.parent_slug,
|
|
)
|
|
continue
|
|
|
|
resolved_roles = resolve_roles(link_seed.required_roles)
|
|
|
|
payload = {
|
|
"group_id": group_id,
|
|
"parent_link_id": parent_id,
|
|
"slug": link_seed.slug,
|
|
"label": link_seed.label,
|
|
"route_name": link_seed.route_name,
|
|
"href_override": link_seed.href_override,
|
|
"match_prefix": link_seed.match_prefix,
|
|
"sort_order": link_seed.sort_order,
|
|
"icon": link_seed.icon,
|
|
"tooltip": link_seed.tooltip,
|
|
"required_roles": resolved_roles,
|
|
"is_enabled": link_seed.is_enabled,
|
|
"is_external": link_seed.is_external,
|
|
}
|
|
conn.execute(link_insert_sql, payload)
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT id FROM navigation_links WHERE group_id = :group_id AND slug = :slug"
|
|
),
|
|
{"group_id": group_id, "slug": link_seed.slug},
|
|
).fetchone()
|
|
if row is not None:
|
|
link_ids[(link_seed.group_slug, link_seed.slug)] = row.id
|
|
|
|
|
|
def _project_id_by_name(conn, project_name: str) -> Optional[int]:
|
|
row = conn.execute(
|
|
text("SELECT id FROM projects WHERE name = :name"),
|
|
{"name": project_name},
|
|
).fetchone()
|
|
return row.id if row else None
|
|
|
|
|
|
def ensure_default_projects(engine: Engine, is_sqlite: bool) -> None:
|
|
with engine.begin() as conn:
|
|
for project in DEFAULT_PROJECTS:
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO projects (name, location, operation_type, description)
|
|
VALUES (:name, :location, :operation_type, :description)
|
|
ON CONFLICT (name) DO UPDATE SET
|
|
location = EXCLUDED.location,
|
|
operation_type = EXCLUDED.operation_type,
|
|
description = EXCLUDED.description
|
|
"""
|
|
),
|
|
project.model_dump(),
|
|
)
|
|
|
|
|
|
def ensure_default_scenarios(engine: Engine, is_sqlite: bool) -> None:
|
|
with engine.begin() as conn:
|
|
for scenario in DEFAULT_SCENARIOS:
|
|
project_id = _project_id_by_name(conn, scenario.project_name)
|
|
if project_id is None:
|
|
logger.warning(
|
|
"Skipping scenario seed '%s' because project '%s' does not exist",
|
|
scenario.name,
|
|
scenario.project_name,
|
|
)
|
|
continue
|
|
|
|
payload = scenario.model_dump(exclude={"project_name"})
|
|
payload.update({"project_id": project_id})
|
|
if is_sqlite:
|
|
sql = """
|
|
INSERT INTO scenarios (
|
|
project_id, name, description, status, discount_rate,
|
|
currency, primary_resource
|
|
)
|
|
VALUES (
|
|
:project_id, :name, :description, :status,
|
|
:discount_rate, :currency, :primary_resource
|
|
)
|
|
ON CONFLICT (project_id, name) DO UPDATE SET
|
|
description = EXCLUDED.description,
|
|
status = EXCLUDED.status,
|
|
discount_rate = EXCLUDED.discount_rate,
|
|
currency = EXCLUDED.currency,
|
|
primary_resource = EXCLUDED.primary_resource
|
|
"""
|
|
else:
|
|
sql = """
|
|
INSERT INTO scenarios (
|
|
project_id, name, description, status, discount_rate,
|
|
currency, primary_resource
|
|
)
|
|
VALUES (
|
|
:project_id, :name, :description, CAST(:status AS scenariostatus),
|
|
:discount_rate, :currency,
|
|
CASE WHEN :primary_resource IS NULL
|
|
THEN NULL
|
|
ELSE CAST(:primary_resource AS resourcetype)
|
|
END
|
|
)
|
|
ON CONFLICT (project_id, name) DO UPDATE SET
|
|
description = EXCLUDED.description,
|
|
status = EXCLUDED.status,
|
|
discount_rate = EXCLUDED.discount_rate,
|
|
currency = EXCLUDED.currency,
|
|
primary_resource = EXCLUDED.primary_resource
|
|
"""
|
|
conn.execute(text(sql), payload)
|
|
|
|
|
|
def ensure_default_financial_inputs(engine: Engine, is_sqlite: bool) -> None:
|
|
with engine.begin() as conn:
|
|
for item in DEFAULT_FINANCIAL_INPUTS:
|
|
project_id = _project_id_by_name(conn, item.project_name)
|
|
if project_id is None:
|
|
logger.warning(
|
|
"Skipping financial input '%s'; project '%s' missing",
|
|
item.name,
|
|
item.project_name,
|
|
)
|
|
continue
|
|
|
|
scenario_row = conn.execute(
|
|
text(
|
|
"SELECT id FROM scenarios WHERE project_id = :project_id AND name = :name"
|
|
),
|
|
{"project_id": project_id, "name": item.scenario_name},
|
|
).fetchone()
|
|
if scenario_row is None:
|
|
logger.warning(
|
|
"Skipping financial input '%s'; scenario '%s' missing for project '%s'",
|
|
item.name,
|
|
item.scenario_name,
|
|
item.project_name,
|
|
)
|
|
continue
|
|
|
|
payload = item.model_dump(
|
|
exclude={"project_name", "scenario_name"},
|
|
)
|
|
if is_sqlite:
|
|
# Convert Decimal to float for SQLite
|
|
payload["amount"] = float(payload["amount"])
|
|
payload.update({"scenario_id": scenario_row.id})
|
|
if is_sqlite:
|
|
sql = """
|
|
INSERT INTO financial_inputs (
|
|
scenario_id, name, category, cost_bucket, amount, currency, notes
|
|
)
|
|
VALUES (
|
|
:scenario_id, :name, :category, :cost_bucket,
|
|
:amount, :currency, :notes
|
|
)
|
|
ON CONFLICT (scenario_id, name) DO UPDATE SET
|
|
category = EXCLUDED.category,
|
|
cost_bucket = EXCLUDED.cost_bucket,
|
|
amount = EXCLUDED.amount,
|
|
currency = EXCLUDED.currency,
|
|
notes = EXCLUDED.notes
|
|
"""
|
|
else:
|
|
sql = """
|
|
INSERT INTO financial_inputs (
|
|
scenario_id, name, category, cost_bucket, amount, currency, notes
|
|
)
|
|
VALUES (
|
|
:scenario_id, :name, CAST(:category AS financialcategory),
|
|
CASE WHEN :cost_bucket IS NULL THEN NULL
|
|
ELSE CAST(:cost_bucket AS costbucket)
|
|
END,
|
|
:amount,
|
|
:currency,
|
|
:notes
|
|
)
|
|
ON CONFLICT (scenario_id, name) DO UPDATE SET
|
|
category = EXCLUDED.category,
|
|
cost_bucket = EXCLUDED.cost_bucket,
|
|
amount = EXCLUDED.amount,
|
|
currency = EXCLUDED.currency,
|
|
notes = EXCLUDED.notes
|
|
"""
|
|
conn.execute(text(sql), payload)
|
|
|
|
|
|
def init_db(database_url: Optional[str] = None) -> None:
|
|
"""Run the idempotent initialization sequence.
|
|
|
|
Steps:
|
|
- Ensure enum types exist.
|
|
- Ensure required tables exist.
|
|
- Seed roles and admin user.
|
|
- Ensure default pricing settings record exists.
|
|
- Seed sample projects, scenarios, and financial inputs.
|
|
"""
|
|
database_url = database_url or _get_database_url()
|
|
is_sqlite = _is_sqlite(database_url)
|
|
engine = _create_engine(database_url)
|
|
logger.info("Starting DB initialization using engine=%s", engine)
|
|
ensure_enums(engine, is_sqlite)
|
|
normalize_enum_values(engine, is_sqlite)
|
|
ensure_tables(engine, is_sqlite)
|
|
ensure_constraints(engine, is_sqlite)
|
|
seed_roles(engine, is_sqlite)
|
|
seed_admin_user(engine, is_sqlite)
|
|
ensure_default_pricing(engine, is_sqlite)
|
|
seed_navigation(engine, is_sqlite)
|
|
ensure_default_projects(engine, is_sqlite)
|
|
ensure_default_scenarios(engine, is_sqlite)
|
|
ensure_default_financial_inputs(engine, is_sqlite)
|
|
logger.info("DB initialization complete")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Allow running manually: python -m scripts.init_db
|
|
logging.basicConfig(level=logging.INFO)
|
|
init_db()
|