settings.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135
  1. from fastapi import APIRouter, Depends
  2. from sqlalchemy.ext.asyncio import AsyncSession
  3. from sqlalchemy import select
  4. from backend.app.core.database import get_db
  5. from backend.app.models.settings import Settings
  6. from backend.app.schemas.settings import AppSettings, AppSettingsUpdate
  7. router = APIRouter(prefix="/settings", tags=["settings"])
  8. # Default settings
  9. DEFAULT_SETTINGS = AppSettings()
  10. async def get_setting(db: AsyncSession, key: str) -> str | None:
  11. """Get a single setting value by key."""
  12. result = await db.execute(select(Settings).where(Settings.key == key))
  13. setting = result.scalar_one_or_none()
  14. return setting.value if setting else None
  15. async def set_setting(db: AsyncSession, key: str, value: str) -> None:
  16. """Set a single setting value."""
  17. result = await db.execute(select(Settings).where(Settings.key == key))
  18. setting = result.scalar_one_or_none()
  19. if setting:
  20. setting.value = value
  21. else:
  22. setting = Settings(key=key, value=value)
  23. db.add(setting)
  24. @router.get("/", response_model=AppSettings)
  25. async def get_settings(db: AsyncSession = Depends(get_db)):
  26. """Get all application settings."""
  27. settings_dict = DEFAULT_SETTINGS.model_dump()
  28. # Load saved settings from database
  29. result = await db.execute(select(Settings))
  30. db_settings = result.scalars().all()
  31. for setting in db_settings:
  32. if setting.key in settings_dict:
  33. # Parse the value based on the expected type
  34. if setting.key in ["auto_archive", "save_thumbnails", "capture_finish_photo", "spoolman_enabled", "check_updates"]:
  35. settings_dict[setting.key] = setting.value.lower() == "true"
  36. elif setting.key in ["default_filament_cost", "energy_cost_per_kwh"]:
  37. settings_dict[setting.key] = float(setting.value)
  38. else:
  39. settings_dict[setting.key] = setting.value
  40. return AppSettings(**settings_dict)
  41. @router.put("/", response_model=AppSettings)
  42. async def update_settings(
  43. settings_update: AppSettingsUpdate,
  44. db: AsyncSession = Depends(get_db),
  45. ):
  46. """Update application settings."""
  47. update_data = settings_update.model_dump(exclude_unset=True)
  48. for key, value in update_data.items():
  49. # Convert value to string for storage
  50. if isinstance(value, bool):
  51. str_value = "true" if value else "false"
  52. else:
  53. str_value = str(value)
  54. await set_setting(db, key, str_value)
  55. await db.commit()
  56. # Return updated settings
  57. return await get_settings(db)
  58. @router.post("/reset", response_model=AppSettings)
  59. async def reset_settings(db: AsyncSession = Depends(get_db)):
  60. """Reset all settings to defaults."""
  61. # Delete all settings
  62. result = await db.execute(select(Settings))
  63. for setting in result.scalars().all():
  64. await db.delete(setting)
  65. await db.commit()
  66. return DEFAULT_SETTINGS
  67. @router.get("/check-ffmpeg")
  68. async def check_ffmpeg():
  69. """Check if ffmpeg is installed and available."""
  70. from backend.app.services.camera import get_ffmpeg_path
  71. ffmpeg_path = get_ffmpeg_path()
  72. return {
  73. "installed": ffmpeg_path is not None,
  74. "path": ffmpeg_path,
  75. }
  76. @router.get("/spoolman")
  77. async def get_spoolman_settings(db: AsyncSession = Depends(get_db)):
  78. """Get Spoolman integration settings."""
  79. spoolman_enabled = await get_setting(db, "spoolman_enabled") or "false"
  80. spoolman_url = await get_setting(db, "spoolman_url") or ""
  81. spoolman_sync_mode = await get_setting(db, "spoolman_sync_mode") or "auto"
  82. return {
  83. "spoolman_enabled": spoolman_enabled,
  84. "spoolman_url": spoolman_url,
  85. "spoolman_sync_mode": spoolman_sync_mode,
  86. }
  87. @router.put("/spoolman")
  88. async def update_spoolman_settings(
  89. settings: dict,
  90. db: AsyncSession = Depends(get_db),
  91. ):
  92. """Update Spoolman integration settings."""
  93. if "spoolman_enabled" in settings:
  94. await set_setting(db, "spoolman_enabled", settings["spoolman_enabled"])
  95. if "spoolman_url" in settings:
  96. await set_setting(db, "spoolman_url", settings["spoolman_url"])
  97. if "spoolman_sync_mode" in settings:
  98. await set_setting(db, "spoolman_sync_mode", settings["spoolman_sync_mode"])
  99. await db.commit()
  100. # Return updated settings
  101. return await get_spoolman_settings(db)