mi
128 lines
5.2 KiB
Python
128 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from sqlalchemy import func, or_
|
|
from sqlalchemy.dialects.postgresql import insert
|
|
|
|
from app.chat_settings import CHAT_MESSAGE_MAX_LENGTH_KEY, validate_chat_settings
|
|
from app.db import AppSetting, Database
|
|
from app.otp_settings import OTP_SETTING_KEYS, validate_otp_settings
|
|
from app.settings import get_settings
|
|
|
|
VALUE_TYPES = {"boolean", "integer", "string", "string_list"}
|
|
|
|
|
|
def load_seed(path: Path) -> list[dict[str, Any]]:
|
|
document = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
if not isinstance(document, dict) or document.get("schema_version") != 1:
|
|
raise ValueError("settings file must have schema_version: 1")
|
|
settings = document.get("settings")
|
|
if not isinstance(settings, dict) or not settings:
|
|
raise ValueError("settings file must contain a non-empty settings mapping")
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
for key, raw in settings.items():
|
|
if not isinstance(key, str) or not key.strip():
|
|
raise ValueError("setting keys must be non-empty strings")
|
|
if not isinstance(raw, dict):
|
|
raise ValueError(f"{key}: setting must be a mapping")
|
|
value_type = raw.get("type")
|
|
if value_type not in VALUE_TYPES:
|
|
raise ValueError(f"{key}: unsupported type {value_type!r}")
|
|
if key in OTP_SETTING_KEYS and value_type != "integer":
|
|
raise ValueError(f"{key}: type must be integer")
|
|
if key == CHAT_MESSAGE_MAX_LENGTH_KEY and value_type != "integer":
|
|
raise ValueError(f"{key}: type must be integer")
|
|
if not isinstance(raw.get("public"), bool):
|
|
raise ValueError(f"{key}: public must be a boolean")
|
|
if key in OTP_SETTING_KEYS and raw["public"]:
|
|
raise ValueError(f"{key}: OTP setting must not be public")
|
|
if key == CHAT_MESSAGE_MAX_LENGTH_KEY and not raw["public"]:
|
|
raise ValueError(f"{key}: setting must be public")
|
|
description = raw.get("description")
|
|
if description is not None and not isinstance(description, str):
|
|
raise ValueError(f"{key}: description must be a string")
|
|
rows.append(
|
|
{
|
|
"setting_key": key,
|
|
"setting_value": serialize_value(key, value_type, raw.get("value")),
|
|
"value_type": value_type,
|
|
"is_public": raw["public"],
|
|
"description": description,
|
|
"record_status": "A",
|
|
}
|
|
)
|
|
validate_otp_settings({row["setting_key"]: row["setting_value"] for row in rows})
|
|
validate_chat_settings({row["setting_key"]: row["setting_value"] for row in rows})
|
|
return rows
|
|
|
|
|
|
def serialize_value(key: str, value_type: str, value: Any) -> str:
|
|
if value_type == "boolean":
|
|
if not isinstance(value, bool):
|
|
raise ValueError(f"{key}: boolean value expected")
|
|
return str(value).lower()
|
|
if value_type == "integer":
|
|
if not isinstance(value, int) or isinstance(value, bool):
|
|
raise ValueError(f"{key}: integer value expected")
|
|
return str(value)
|
|
if value_type == "string_list":
|
|
if isinstance(value, list) and all(isinstance(item, str) for item in value):
|
|
return ",".join(value)
|
|
if isinstance(value, str):
|
|
return value
|
|
raise ValueError(f"{key}: string or list of strings expected")
|
|
if not isinstance(value, str):
|
|
raise ValueError(f"{key}: string value expected")
|
|
return value
|
|
|
|
|
|
async def seed(path: Path) -> int:
|
|
rows = load_seed(path)
|
|
database = Database(get_settings().database_url)
|
|
try:
|
|
async with database.sessions() as session:
|
|
for row in rows:
|
|
statement = insert(AppSetting).values(**row)
|
|
excluded = statement.excluded
|
|
statement = statement.on_conflict_do_update(
|
|
index_elements=[AppSetting.setting_key],
|
|
set_={
|
|
"setting_value": excluded.setting_value,
|
|
"value_type": excluded.value_type,
|
|
"is_public": excluded.is_public,
|
|
"description": excluded.description,
|
|
"record_status": "A",
|
|
"updated_at": func.now(),
|
|
},
|
|
where=or_(
|
|
AppSetting.setting_value.is_distinct_from(excluded.setting_value),
|
|
AppSetting.value_type.is_distinct_from(excluded.value_type),
|
|
AppSetting.is_public.is_distinct_from(excluded.is_public),
|
|
AppSetting.description.is_distinct_from(excluded.description),
|
|
AppSetting.record_status != "A",
|
|
),
|
|
)
|
|
await session.execute(statement)
|
|
await session.commit()
|
|
finally:
|
|
await database.close()
|
|
return len(rows)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Idempotently seed application settings")
|
|
parser.add_argument("--file", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
count = asyncio.run(seed(args.file))
|
|
print(f"Application settings seeded: {count}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|