Madhava/vault and magic import

.gitignore

@@ -169,4 +169,6 @@ data/**
 users/**
 dist
 syftbox.egg-info
-keys/**
+keys/**
+scheduler.lock
+jobs.sqlite

pyproject.toml

@@ -12,6 +12,7 @@ dependencies = [
     "apscheduler>=3.10.4",
     "jinja2>=3.1.4",
     "typing-extensions>=4.12.2",
+    "sqlalchemy>=2.0.34",
 ]
 
 [build-system]

syftbox/client/client.py

@@ -1,14 +1,17 @@
 import argparse
+import atexit
 import importlib
 import os
 import subprocess
 import sys
 import time
+import traceback
 import types
 from dataclasses import dataclass
 from pathlib import Path
 
 import uvicorn
+from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
 from apscheduler.schedulers.background import BackgroundScheduler
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
@@ -260,6 +263,7 @@ def run_plugin(plugin_name):
         module = app.loaded_plugins[plugin_name].module
         module.run(app.shared_state)
     except Exception as e:
+        traceback.print_exc()
         print(e)
 
 
@@ -278,19 +282,25 @@ def start_plugin(plugin_name: str):
 
     try:
         plugin = app.loaded_plugins[plugin_name]
-        job = app.scheduler.add_job(
-            func=run_plugin,
-            trigger="interval",
-            seconds=plugin.schedule / 1000,
-            id=plugin_name,
-            args=[plugin_name],
-        )
-        app.running_plugins[plugin_name] = {
-            "job": job,
-            "start_time": time.time(),
-            "schedule": plugin.schedule,
-        }
-        return {"message": f"Plugin {plugin_name} started successfully"}
+
+        existing_job = app.scheduler.get_job(plugin_name)
+        if existing_job is None:
+            job = app.scheduler.add_job(
+                func=run_plugin,
+                trigger="interval",
+                seconds=plugin.schedule / 1000,
+                id=plugin_name,
+                args=[plugin_name],
+            )
+            app.running_plugins[plugin_name] = {
+                "job": job,
+                "start_time": time.time(),
+                "schedule": plugin.schedule,
+            }
+            return {"message": f"Plugin {plugin_name} started successfully"}
+        else:
+            print(f"Job {existing_job}, already added")
+            return {"message": f"Plugin {plugin_name} already started"}
     except Exception as e:
         raise HTTPException(
             status_code=500,
@@ -316,20 +326,33 @@ def parse_args():
     return parser.parse_args()
 
 
+JOB_FILE = "jobs.sqlite"
+
+
 async def lifespan(app: FastAPI):
     # Startup
     print("> Starting Client")
     args = parse_args()
     client_config = load_or_create_config(args)
     app.shared_state = initialize_shared_state(client_config)
 
-    scheduler = BackgroundScheduler()
+    # Clear the lock file on the first run if it exists
+    if os.path.exists(JOB_FILE):
+        os.remove(JOB_FILE)
+        print(f"> Cleared existing job file: {JOB_FILE}")
+
+    # Start the scheduler
+    jobstores = {"default": SQLAlchemyJobStore(url=f"sqlite:///{JOB_FILE}")}
+    scheduler = BackgroundScheduler(jobstores=jobstores)
     scheduler.start()
     app.scheduler = scheduler
+    atexit.register(stop_scheduler)
+
     app.running_plugins = {}
     app.loaded_plugins = load_plugins(client_config)
 
-    autorun_plugins = ["init", "sync", "create_datasite", "watch_and_run"]
+    # autorun_plugins = ["init", "sync", "create_datasite", "watch_and_run"]
+    autorun_plugins = ["init", "sync", "create_datasite"]
     for plugin in autorun_plugins:
         start_plugin(plugin)
 
@@ -339,6 +362,13 @@ async def lifespan(app: FastAPI):
     print("Shutting down...")
 
 
+def stop_scheduler():
+    # Remove the lock file if it exists
+    if os.path.exists(JOB_FILE):
+        os.remove(JOB_FILE)
+        print("Scheduler stopped and lock file removed.")
+
+
 app = FastAPI(lifespan=lifespan)
 
 app.mount("/static", StaticFiles(directory=current_dir / "static"), name="static")
@@ -504,6 +534,7 @@ def main() -> None:
         port=client_config.port,
         log_level="debug" if debug else "info",
         reload=debug,  # Enable hot reloading only in debug mode
+        reload_dirs="./syftbox",
     )
 
 

syftbox/client/plugins/create_datasite.py

@@ -1,7 +1,7 @@
 import logging
 import os
 
-from syftbox.lib import USER_GROUP_GLOBAL, SyftPermission, perm_file_path
+from syftbox.lib import SyftPermission, perm_file_path
 
 logger = logging.getLogger(__name__)
 
@@ -20,11 +20,7 @@ def claim_datasite(client_config):
     else:
         print(f"> {client_config.email} Creating Datasite + Permfile")
         try:
-            perm_file = SyftPermission(
-                admin=[client_config.email],
-                read=[client_config.email, USER_GROUP_GLOBAL],
-                write=[client_config.email],
-            )
+            perm_file = SyftPermission.datasite_default(client_config.email)
             perm_file.save(file_path)
         except Exception as e:
             print("Failed to create perm file", e)

syftbox/client/plugins/sync.py

@@ -1,4 +1,5 @@
 import os
+import traceback
 from threading import Event
 
 import requests
@@ -432,16 +433,19 @@ def run(shared_state):
                 try:
                     create_datasites(shared_state.client_config)
                 except Exception as e:
+                    traceback.print_exc()
                     print("failed to get_datasites", e)
 
                 try:
                     num_changes += sync_up(shared_state.client_config)
                 except Exception as e:
+                    traceback.print_exc()
                     print("failed to sync up", e)
 
                 try:
                     num_changes += sync_down(shared_state.client_config)
                 except Exception as e:
+                    traceback.print_exc()
                     print("failed to sync down", e)
             if num_changes == 0:
                 print("✅ Synced")