bvandeusenandClaude Fable 5
64c641ce80
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / integration (push) Successful in 25s
CI & Build / TypeScript typecheck (push) Successful in 32s
CI & Build / Python tests (push) Failing after 37s
CI & Build / Build & push image (push) Skipped
Reading the 28 route modules against each other and against the MCP tools: - routes/notes.py carried a PUT and a PATCH handler that were the same function minus the supersedes contract on one of them — one handler now serves both verbs, so both carry it. - The two _attach_supersession copies (REST + MCP) become supersession_svc.attach_relations(uid, note_id, data, hint=) — the seam the two surfaces must agree through; only the agent surface adds the one-sentence reading hint. - Three local _uid() wrappers over g.user.id → scribe.auth.get_current_user_id like every other module; design_systems' private _not_found → routes.utils. not_found; the four "********" literals → settings_svc.SECRET_MASK with the read/write contract written once. - routes/plugin.py: the project_id/repo resolution block and the comma-separated id parse were copied into three endpoints — _project_scope() and _int_list() now. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
106 lines
4.0 KiB
Python
106 lines
4.0 KiB
Python
import logging
|
|
|
|
from sqlalchemy import delete as sa_delete, select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.setting import Setting
|
|
from scribe.models.user import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# What a stored credential looks like on the wire. Every surface that READS a
|
|
# secret (smtp_password, forge_webhook_secret, a forge token) returns this
|
|
# when one is set; every surface that WRITES one treats this value coming back
|
|
# as "unchanged", never as a request to store eight asterisks over the real
|
|
# credential. One constant so the read and write halves cannot disagree.
|
|
SECRET_MASK = "********"
|
|
|
|
|
|
async def get_admin_setting(key: str, default: str = "") -> str:
|
|
"""Read an instance-global setting (one stored on an admin account).
|
|
|
|
Used for settings that aren't per-user — e.g. the plugin marketplace URL
|
|
shown to everyone in Settings. Mirrors the admin-scoped lookup used for
|
|
the application base URL.
|
|
"""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Setting)
|
|
.join(User, Setting.user_id == User.id)
|
|
.where(User.role == "admin", Setting.key == key)
|
|
)
|
|
setting = result.scalars().first()
|
|
return setting.value if setting and setting.value else default
|
|
|
|
|
|
async def set_admin_setting(key: str, value: str) -> None:
|
|
"""Write an instance-global setting onto the first admin account.
|
|
|
|
The write-side counterpart to get_admin_setting, for non-per-user settings
|
|
written outside a request context (e.g. a scheduler persisting its last-run
|
|
summary) where get_current_user_id() isn't available.
|
|
"""
|
|
async with async_session() as session:
|
|
admin_id = (
|
|
await session.execute(
|
|
select(User.id).where(User.role == "admin").order_by(User.id)
|
|
)
|
|
).scalars().first()
|
|
if admin_id is not None:
|
|
await set_setting(admin_id, key, value)
|
|
|
|
|
|
async def get_setting(user_id: int, key: str, default: str = "") -> str:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Setting).where(Setting.user_id == user_id, Setting.key == key)
|
|
)
|
|
setting = result.scalar_one_or_none()
|
|
return setting.value if setting else default
|
|
|
|
|
|
async def set_setting(user_id: int, key: str, value: str) -> None:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Setting).where(Setting.user_id == user_id, Setting.key == key)
|
|
)
|
|
setting = result.scalar_one_or_none()
|
|
if setting:
|
|
setting.value = value
|
|
else:
|
|
session.add(Setting(user_id=user_id, key=key, value=value))
|
|
await session.commit()
|
|
|
|
|
|
async def set_settings_batch(user_id: int, settings: dict[str, str]) -> None:
|
|
"""Update multiple settings in a single transaction."""
|
|
async with async_session() as session:
|
|
for key, value in settings.items():
|
|
result = await session.execute(
|
|
select(Setting).where(Setting.user_id == user_id, Setting.key == key)
|
|
)
|
|
setting = result.scalar_one_or_none()
|
|
if setting:
|
|
setting.value = value
|
|
else:
|
|
session.add(Setting(user_id=user_id, key=key, value=value))
|
|
await session.commit()
|
|
logger.info("Batch-updated %d settings for user %d", len(settings), user_id)
|
|
|
|
|
|
async def delete_setting(user_id: int, key: str) -> None:
|
|
"""Remove a setting row so get_setting() returns its hardcoded default instead."""
|
|
async with async_session() as session:
|
|
await session.execute(
|
|
sa_delete(Setting).where(Setting.user_id == user_id, Setting.key == key)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def get_all_settings(user_id: int) -> dict[str, str]:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Setting).where(Setting.user_id == user_id)
|
|
)
|
|
return {s.key: s.value for s in result.scalars().all()}
|