v2.16.1

Merge pull request #1859 from AntaresSimulatorTeam/release/2.16.1

antarest/__init__.py

@@ -7,9 +7,9 @@
 
 # Standard project metadata
 
-__version__ = "2.16.0"
+__version__ = "2.16.1"
 __author__ = "RTE, Antares Web Team"
-__date__ = "2023-11-30"
+__date__ = "2023-12-14"
 # noinspection SpellCheckingInspection
 __credits__ = "(c) Réseau de Transport de l’Électricité (RTE)"
 

antarest/core/exceptions.py

@@ -189,6 +189,11 @@ def __init__(self, message: str) -> None:
         super().__init__(HTTPStatus.NOT_FOUND, message)
 
 
+class DuplicateConstraintName(HTTPException):
+    def __init__(self, message: str) -> None:
+        super().__init__(HTTPStatus.CONFLICT, message)
+
+
 class MissingDataError(HTTPException):
     def __init__(self, message: str) -> None:
         super().__init__(HTTPStatus.NOT_FOUND, message)

antarest/core/jwt.py

@@ -3,9 +3,7 @@
 from pydantic import BaseModel
 
 from antarest.core.roles import RoleType
-from antarest.login.model import Group, Identity
-
-ADMIN_ID = 1
+from antarest.login.model import ADMIN_ID, Group, Identity
 
 
 class JWTGroup(BaseModel):

antarest/core/tasks/model.py

@@ -1,11 +1,12 @@
 import uuid
 from datetime import datetime
 from enum import Enum
-from typing import Any, List, Optional
+from typing import Any, List, Mapping, Optional
 
 from pydantic import BaseModel, Extra
 from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, Sequence, String  # type: ignore
-from sqlalchemy.orm import relationship  # type: ignore
+from sqlalchemy.engine.base import Engine  # type: ignore
+from sqlalchemy.orm import Session, relationship, sessionmaker  # type: ignore
 
 from antarest.core.persistence import Base
 
@@ -171,3 +172,29 @@ def __repr__(self) -> str:
             f" result_msg={self.result_msg},"
             f" result_status={self.result_status}"
         )
+
+
+def cancel_orphan_tasks(engine: Engine, session_args: Mapping[str, bool]) -> None:
+    """
+    Cancel all tasks that are currently running or pending.
+
+    When the web application restarts, such as after a new deployment, any pending or running tasks may be lost.
+    To mitigate this, it is preferable to set these tasks to a "FAILED" status.
+    This ensures that users can easily identify the tasks that were affected by the restart and take appropriate
+    actions, such as restarting the tasks manually.
+
+    Args:
+        engine: The database engine (SQLAlchemy connection to SQLite or PostgreSQL).
+        session_args: The session arguments (SQLAlchemy session arguments).
+    """
+    updated_values = {
+        TaskJob.status: TaskStatus.FAILED.value,
+        TaskJob.result_status: False,
+        TaskJob.result_msg: "Task was interrupted due to server restart",
+        TaskJob.completion_date: datetime.utcnow(),
+    }
+    with sessionmaker(bind=engine, **session_args)() as session:
+        session.query(TaskJob).filter(TaskJob.status.in_([TaskStatus.RUNNING.value, TaskStatus.PENDING.value])).update(
+            updated_values, synchronize_session=False
+        )
+        session.commit()

antarest/core/tasks/repository.py

@@ -1,26 +1,56 @@
 import datetime
+import typing as t
 from http import HTTPStatus
 from operator import and_
-from typing import Any, List, Optional
 
 from fastapi import HTTPException
+from sqlalchemy.orm import Session  # type: ignore
 
 from antarest.core.tasks.model import TaskJob, TaskListFilter, TaskStatus
 from antarest.core.utils.fastapi_sqlalchemy import db
 from antarest.core.utils.utils import assert_this
 
 
 class TaskJobRepository:
+    """
+    Database connector to manage Tasks/Jobs entities.
+    """
+
+    def __init__(self, session: t.Optional[Session] = None):
+        """
+        Initialize the repository.
+
+        Args:
+            session: Optional SQLAlchemy session to be used.
+        """
+        self._session = session
+
+    @property
+    def session(self) -> Session:
+        """
+        Get the SQLAlchemy session for the repository.
+
+        Returns:
+            SQLAlchemy session.
+        """
+        if self._session is None:
+            # Get or create the session from a context variable (thread local variable)
+            return db.session
+        # Get the user-defined session
+        return self._session
+
     def save(self, task: TaskJob) -> TaskJob:
-        task = db.session.merge(task)
-        db.session.add(task)
-        db.session.commit()
+        session = self.session
+        task = session.merge(task)
+        session.add(task)
+        session.commit()
         return task
 
-    def get(self, id: str) -> Optional[TaskJob]:
-        task: TaskJob = db.session.get(TaskJob, id)
+    def get(self, id: str) -> t.Optional[TaskJob]:
+        session = self.session
+        task: TaskJob = session.get(TaskJob, id)
         if task is not None:
-            db.session.refresh(task)
+            session.refresh(task)
         return task
 
     def get_or_raise(self, id: str) -> TaskJob:
@@ -30,7 +60,7 @@ def get_or_raise(self, id: str) -> TaskJob:
         return task
 
     @staticmethod
-    def _combine_clauses(where_clauses: List[Any]) -> Any:
+    def _combine_clauses(where_clauses: t.List[t.Any]) -> t.Any:
         assert_this(len(where_clauses) > 0)
         if len(where_clauses) > 1:
             return and_(
@@ -40,9 +70,9 @@ def _combine_clauses(where_clauses: List[Any]) -> Any:
         else:
             return where_clauses[0]
 
-    def list(self, filter: TaskListFilter, user: Optional[int] = None) -> List[TaskJob]:
-        query = db.session.query(TaskJob)
-        where_clauses: List[Any] = []
+    def list(self, filter: TaskListFilter, user: t.Optional[int] = None) -> t.List[TaskJob]:
+        query = self.session.query(TaskJob)
+        where_clauses: t.List[t.Any] = []
         if user:
             where_clauses.append(TaskJob.owner_id == user)
         if len(filter.status) > 0:
@@ -74,19 +104,21 @@ def list(self, filter: TaskListFilter, user: Optional[int] = None) -> List[TaskJ
         elif len(where_clauses) == 1:
             query = query.where(*where_clauses)
 
-        tasks: List[TaskJob] = query.all()
+        tasks: t.List[TaskJob] = query.all()
         return tasks
 
     def delete(self, tid: str) -> None:
-        task = db.session.get(TaskJob, tid)
+        session = self.session
+        task = session.get(TaskJob, tid)
         if task:
-            db.session.delete(task)
-            db.session.commit()
+            session.delete(task)
+            session.commit()
 
     def update_timeout(self, task_id: str, timeout: int) -> None:
         """Update task status to TIMEOUT."""
-        task: TaskJob = db.session.get(TaskJob, task_id)
+        session = self.session
+        task: TaskJob = session.get(TaskJob, task_id)
         task.status = TaskStatus.TIMEOUT
         task.result_msg = f"Task '{task_id}' timeout after {timeout} seconds"
         task.result_status = False
-        db.session.commit()
+        session.commit()