Merge pull request #673 from fractal-analytics-platform/fix_sqlite_ta…

…sk_collection

Fix task collection with sqlite

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 # 1.2.5
 
-* Fix bug in task collection when using sqlite (\#664).
+* Fix bug in task collection when using sqlite (\#664, \#673).
 * Fix bug in task collection from local package, where package extras were not considered (\#671).
 * Improve error handling in workflow-apply endpoint (\#665).
 * Fix a bug upon project removal in the presence of project-related jobs (\#666). Note: this removes the `ApplyWorkflow.Project` attribute.

fractal_server/__init__.py

@@ -1 +1 @@
-__VERSION__ = "1.2.5a0"
+__VERSION__ = "1.2.5a2"

fractal_server/app/api/v1/task.py

@@ -36,7 +36,9 @@
 from ....tasks.collection import get_log_path
 from ....tasks.collection import inspect_package
 from ...db import AsyncSession
+from ...db import DBSyncSession
 from ...db import get_db
+from ...db import get_sync_db
 from ...models import State
 from ...models import Task
 from ...security import current_active_superuser
@@ -61,9 +63,9 @@ async def _background_collect_pip(
     directory.
     """
 
-    async for db in get_db():
+    with next(get_sync_db()) as db:
 
-        state: State = await db.get(State, state_id)
+        state: State = db.get(State, state_id)
 
         logger_name = task_pkg.package.replace("/", "_")
         logger = set_logger(
@@ -81,8 +83,8 @@ async def _background_collect_pip(
             data.status = "installing"
 
             state.data = data.sanitised_dict()
-            await db.merge(state)
-            await db.commit()
+            db.merge(state)
+            db.commit()
             task_list = await create_package_environment_pip(
                 venv_path=venv_path,
                 task_pkg=task_pkg,
@@ -93,8 +95,8 @@ async def _background_collect_pip(
             logger.debug("Task-collection status: collecting")
             data.status = "collecting"
             state.data = data.sanitised_dict()
-            await db.merge(state)
-            await db.commit()
+            db.merge(state)
+            db.commit()
             tasks = await _insert_tasks(task_list=task_list, db=db)
 
             # finalise
@@ -109,14 +111,14 @@ async def _background_collect_pip(
             data.log = get_collection_log(venv_path)
             state.data = data.sanitised_dict()
             db.add(state)
-            await db.merge(state)
-            await db.commit()
+            db.merge(state)
+            db.commit()
 
             # Write last logs to file
             logger.debug("Task-collection status: OK")
             logger.info("Background task collection completed successfully")
             close_logger(logger)
-            await db.close()
+            db.close()
 
         except Exception as e:
             # Write last logs to file
@@ -129,26 +131,27 @@ async def _background_collect_pip(
             data.info = f"Original error: {e}"
             data.log = get_collection_log(venv_path)
             state.data = data.sanitised_dict()
-            await db.merge(state)
-            await db.commit()
-            await db.close()
+            db.merge(state)
+            db.commit()
+            db.close()
 
             # Delete corrupted package dir
             shell_rmtree(venv_path)
 
 
 async def _insert_tasks(
     task_list: list[TaskCreate],
-    db: AsyncSession,
+    db: DBSyncSession,
 ) -> list[Task]:
     """
     Insert tasks into database
     """
     task_db_list = [Task.from_orm(t) for t in task_list]
     db.add_all(task_db_list)
-    await db.commit()
-    await asyncio.gather(*[db.refresh(t) for t in task_db_list])
-    await db.close()
+    db.commit()
+    for t in task_db_list:
+        db.refresh(t)
+    db.close()
     return task_db_list
 
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "fractal-server"
-version = "1.2.5a0"
+version = "1.2.5a2"
 description = "Server component of the Fractal analytics platform"
 authors = [
     "Jacopo Nespolo <[email protected]>",
@@ -75,7 +75,7 @@ filterwarnings = [
 ]
 
 [tool.bumpver]
-current_version = "1.2.5a0"
+current_version = "1.2.5a2"
 version_pattern = "MAJOR.MINOR.PATCH[PYTAGNUM]"
 commit_message = "bump version {old_version} -> {new_version}"
 commit = true

tests/fixtures_tasks.py

@@ -168,10 +168,10 @@ async def install_dummy_packages(tmp777_session_path, dummy_task_package):
 
 
 @pytest.fixture(scope="function")
-async def collect_packages(db, install_dummy_packages):
+async def collect_packages(db_sync, install_dummy_packages):
     from fractal_server.app.api.v1.task import _insert_tasks
 
-    tasks = await _insert_tasks(task_list=install_dummy_packages, db=db)
+    tasks = await _insert_tasks(task_list=install_dummy_packages, db=db_sync)
     return tasks