tonym/spoolman2
1
0
Fork 0
Code Issues 19 Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Added logging for when scheduled backup runs

Browse Source
This commit is contained in:
Donkie
2023-07-07 17:37:17 +02:00
parent 66b6770ddd
commit d3a4dcfcf8
2 changed files with 12 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spoolman/api/v1/router.py
View File

@@ -9,7 +9,7 @@ from fastapi.responses import JSONResponse
from starlette.requests import Request
from starlette.responses import Response
from spoolman.database.database import backup_task
from spoolman.database.database import backup_global_db
from spoolman.exceptions import ItemNotFoundError
from . import filament, models, spool, vendor
@@ -48,7 +48,7 @@ async def health() -> models.HealthCheck:
)
async def backup(): # noqa: ANN201
"""Trigger a database backup."""
path = await backup_task()
path = await backup_global_db()
if path is None:
return JSONResponse(
status_code=500,

12
spoolman/database/database.py
View File

@@ -164,12 +164,20 @@ def setup_db(connection_url: URL) -> None:
__db.connect()
async def backup_task() -> Optional[Path]:
async def backup_global_db(num_backups: int = 5) -> Optional[Path]:
"""Backup the database and rotate existing backups.
Returns:
The path to the created backup or None if no backup was created.
"""
if __db is None:
raise RuntimeError("DB is not setup.")
return __db.backup_and_rotate(env.get_data_dir().joinpath("backups"), num_backups=num_backups)
async def _backup_task() -> Optional[Path]:
"""Perform scheduled backup of the database."""
logger.info("Performing scheduled database backup.")
if __db is None:
raise RuntimeError("DB is not setup.")
return __db.backup_and_rotate(env.get_data_dir().joinpath("backups"), num_backups=5)
@@ -188,7 +196,7 @@ def schedule_tasks(scheduler: Scheduler) -> None:
if "sqlite" in __db.connection_url.drivername:
logger.info("Scheduling automatic database backup for midnight.")
# Schedule for midnight
scheduler.daily(datetime.time(hour=0, minute=0, second=0), backup_task) # type: ignore[arg-type]
scheduler.daily(datetime.time(hour=0, minute=0, second=0), _backup_task) # type: ignore[arg-type]
async def get_db_session() -> AsyncGenerator[AsyncSession, None]: