121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.admin.service import ensure_default_roles, get_or_create_account
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.db.models import Role, SystemRoleAssignment, Tenant, User, UserRoleAssignment
|
|
from govoplan_core.db.session import get_database
|
|
from govoplan_core.security.api_keys import CreatedApiKey, create_api_key
|
|
from govoplan_core.security.permissions import TENANT_SCOPES
|
|
|
|
DEFAULT_SCOPES = sorted(TENANT_SCOPES)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class BootstrapResult:
|
|
tenant: Tenant
|
|
user: User
|
|
created_api_key: CreatedApiKey | None
|
|
|
|
|
|
def create_all_tables() -> None:
|
|
# Import models so SQLAlchemy sees all tables.
|
|
from govoplan_core.db import models # noqa: F401
|
|
from govoplan_core.access.db import models as access_models # noqa: F401
|
|
from govoplan_files.backend.db import models as file_models # noqa: F401
|
|
from govoplan_mail.backend.db import models as mail_models # noqa: F401
|
|
from govoplan_campaign.backend.db import models as campaign_models # noqa: F401
|
|
from govoplan_core.access.db.base import AccessBase
|
|
|
|
engine = get_database().engine
|
|
Base.metadata.create_all(bind=engine)
|
|
AccessBase.metadata.create_all(bind=engine)
|
|
|
|
|
|
def bootstrap_dev_data(
|
|
session: Session,
|
|
*,
|
|
api_key_secret: str | None = None,
|
|
tenant_slug: str = "default",
|
|
user_email: str = "admin@example.local",
|
|
user_password: str = "dev-admin",
|
|
) -> BootstrapResult:
|
|
tenant = session.query(Tenant).filter(Tenant.slug == tenant_slug).one_or_none()
|
|
if tenant is None:
|
|
tenant = Tenant(slug=tenant_slug, name="Default Tenant", default_locale="en", settings={})
|
|
session.add(tenant)
|
|
session.flush()
|
|
|
|
tenant_roles = ensure_default_roles(session, tenant)
|
|
system_roles = ensure_default_roles(session, None)
|
|
|
|
account, _, _ = get_or_create_account(
|
|
session,
|
|
email=user_email,
|
|
display_name="Development Admin",
|
|
password=user_password,
|
|
password_reset_required=False,
|
|
)
|
|
# Keep development credentials deterministic when an old create_all database
|
|
# is reused. This is guarded by the explicit dev bootstrap setting.
|
|
from govoplan_core.security.passwords import hash_password
|
|
|
|
if not account.password_hash:
|
|
account.password_hash = hash_password(user_password)
|
|
account.is_active = True
|
|
session.add(account)
|
|
|
|
user = session.query(User).filter(User.tenant_id == tenant.id, User.account_id == account.id).one_or_none()
|
|
if user is None:
|
|
user = User(
|
|
tenant_id=tenant.id,
|
|
account_id=account.id,
|
|
email=account.email,
|
|
display_name="Development Admin",
|
|
is_tenant_admin=True, # legacy presentation flag; RBAC uses the owner role below.
|
|
auth_provider=account.auth_provider,
|
|
password_hash=account.password_hash,
|
|
)
|
|
session.add(user)
|
|
session.flush()
|
|
else:
|
|
user.email = account.email
|
|
user.password_hash = account.password_hash
|
|
user.is_active = True
|
|
session.add(user)
|
|
|
|
owner_role = tenant_roles["owner"]
|
|
existing_assignment = session.query(UserRoleAssignment).filter(
|
|
UserRoleAssignment.tenant_id == tenant.id,
|
|
UserRoleAssignment.user_id == user.id,
|
|
UserRoleAssignment.role_id == owner_role.id,
|
|
).one_or_none()
|
|
if existing_assignment is None:
|
|
session.add(UserRoleAssignment(tenant_id=tenant.id, user_id=user.id, role_id=owner_role.id))
|
|
|
|
system_owner = system_roles["system_owner"]
|
|
existing_system_assignment = session.query(SystemRoleAssignment).filter(
|
|
SystemRoleAssignment.account_id == account.id,
|
|
SystemRoleAssignment.role_id == system_owner.id,
|
|
).one_or_none()
|
|
if existing_system_assignment is None:
|
|
session.add(SystemRoleAssignment(account_id=account.id, role_id=system_owner.id))
|
|
|
|
created_api_key = None
|
|
if api_key_secret:
|
|
existing = [key for key in user.api_keys if key.name == "Development API key" and key.revoked_at is None]
|
|
if not existing:
|
|
created_api_key = create_api_key(
|
|
session,
|
|
user=user,
|
|
name="Development API key",
|
|
scopes=DEFAULT_SCOPES,
|
|
secret=api_key_secret,
|
|
)
|
|
|
|
session.commit()
|
|
return BootstrapResult(tenant=tenant, user=user, created_api_key=created_api_key)
|