1+ import asyncio
12import logging
2- from typing import Any , Dict , List , Union
3+ import shutil
34
4- from fastapi import APIRouter , Depends , status
5+ from fastapi import APIRouter , Depends , Request , status
56from fastapi .exceptions import HTTPException
67from kink import di , inject
7- from pydantic import BaseModel as PydanticBaseModel
88from sqlalchemy import exc
99from sqlalchemy .orm .session import sessionmaker
1010
11- from DashAI .back .dependencies .database .models import ConverterList , Dataset
11+ from DashAI .back .api .api_v1 .endpoints .jobs import _enqueue_job_logic
12+ from DashAI .back .api .api_v1 .schemas import converter_params as schemas
13+ from DashAI .back .core .enums .status import ConverterListStatus
14+ from DashAI .back .dependencies .database .models import ConverterList , Explorer , Notebook
15+ from DashAI .back .dependencies .job_queues import BaseJobQueue
16+ from DashAI .back .dependencies .job_queues .job_queue import job_queue_loop
17+ from DashAI .back .dependencies .registry import ComponentRegistry
1218
1319logger = logging .getLogger (__name__ )
1420router = APIRouter ()
1521
1622
17- class ConverterParams (PydanticBaseModel ):
18- order : int = 0
19- params : Dict [str , Union [str , int , float , bool , None ]] = None
20- scope : Dict [str , List [int ]] = None
21-
22- def serialize (self ) -> Dict [str , Any ]:
23- return {
24- "order" : self .order ,
25- "params" : self .params ,
26- "scope" : self .scope ,
27- }
28-
29-
30- class ConverterListParams (PydanticBaseModel ):
31- dataset_id : int
32- converters : Dict [str , ConverterParams ]
33-
34-
3523@router .post ("/" , status_code = status .HTTP_201_CREATED )
3624@inject
37- async def post_dataset_converter_list (
38- params : ConverterListParams ,
25+ async def post_notebook_converter_list (
26+ params : schemas . ConverterListParams ,
3927 session_factory : sessionmaker = Depends (lambda : di ["session_factory" ]),
4028):
41- """Save a list of converters to apply to the dataset .
29+ """Save a list of converters to apply to the notebook .
4230
4331 Parameters
4432 ----------
45- dataset_id : int
46- ID of the dataset .
33+ notebook_id : int
34+ ID of the notebook .
4735 converters : Dict[str, ConverterParams]
48- A dictionary with the converters to apply to the dataset .
36+ A dictionary with the converters to apply to the notebook .
4937 session_factory : Callable[..., ContextManager[Session]]
5038 A factory that creates a context manager that handles a SQLAlchemy session.
5139 The generated session can be used to access and query the database.
@@ -58,23 +46,24 @@ async def post_dataset_converter_list(
5846 Raises
5947 ------
6048 HTTPException
61- If the dataset is not found or if there is an internal database error.
49+ If the notebook is not found or if there is an internal database error.
6250 """
6351 with session_factory () as db :
6452 try :
65- dataset = db .get (Dataset , params .dataset_id )
66- if not dataset :
53+ notebook = db .get (Notebook , params .notebook_id )
54+ if not notebook :
6755 raise HTTPException (
6856 status_code = status .HTTP_404_NOT_FOUND ,
69- detail = "Dataset not found" ,
57+ detail = "Notebook not found" ,
7058 )
71- serialized_converters = {
72- key : value . serialize () for key , value in params .converters . items ()
73- }
59+
60+ converter_name = params .converter
61+ converter_parameters = params . parameters . serialize ()
7462
7563 converter_list = ConverterList (
76- dataset_id = params .dataset_id ,
77- converters = serialized_converters ,
64+ notebook_id = params .notebook_id ,
65+ converter = converter_name ,
66+ parameters = converter_parameters ,
7867 )
7968
8069 db .add (converter_list )
@@ -93,7 +82,7 @@ async def post_dataset_converter_list(
9382
9483@router .get ("/{converter_list_id}" )
9584@inject
96- async def get_dataset_converter_list (
85+ async def get_converter_list (
9786 converter_list_id : int ,
9887 session_factory : sessionmaker = Depends (lambda : di ["session_factory" ]),
9988):
@@ -134,3 +123,156 @@ async def get_dataset_converter_list(
134123 status_code = status .HTTP_500_INTERNAL_SERVER_ERROR ,
135124 detail = "Internal database error" ,
136125 ) from e
126+
127+
128+ @router .get ("/notebook/{notebook_id}" )
129+ @inject
130+ async def get_converters_by_notebook (
131+ notebook_id : int ,
132+ session_factory : sessionmaker = Depends (lambda : di ["session_factory" ]),
133+ ):
134+ """Get a list of finished converters from the database by notebook ID.
135+
136+ Parameters
137+ ----------
138+ notebook_id : int
139+ ID of the notebook.
140+ session_factory : Callable[..., ContextManager[Session]]
141+ A factory that creates a context manager that handles a SQLAlchemy session.
142+ The generated session can be used to access and query the database.
143+
144+ Returns
145+ -------
146+ List[ConverterList]
147+ A list of converter lists.
148+
149+ Raises
150+ ------
151+ HTTPException
152+ If there is an internal database error.
153+ """
154+ with session_factory () as db :
155+ try :
156+ converter_lists = (
157+ db .query (ConverterList )
158+ .filter (ConverterList .notebook_id == notebook_id )
159+ .filter (ConverterList .status == ConverterListStatus .FINISHED )
160+ .all ()
161+ )
162+ return converter_lists
163+
164+ except exc .SQLAlchemyError as e :
165+ logger .exception (e )
166+ raise HTTPException (
167+ status_code = status .HTTP_500_INTERNAL_SERVER_ERROR ,
168+ detail = "Internal database error" ,
169+ ) from e
170+
171+
172+ @router .delete ("/{converter_list_id}" )
173+ @inject
174+ async def delete_converter_list (
175+ converter_list_id : int ,
176+ request : Request ,
177+ session_factory : sessionmaker = Depends (lambda : di ["session_factory" ]),
178+ component_registry : ComponentRegistry = Depends (lambda : di ["component_registry" ]),
179+ job_queue : BaseJobQueue = Depends (lambda : di ["job_queue" ]),
180+ ):
181+ """Delete a converter list from the database.
182+
183+ Parameters
184+ ----------
185+ converter_list_id : int
186+ ID of the converter list.
187+ session_factory : Callable[..., ContextManager[Session]]
188+ A factory that creates a context manager that handles a SQLAlchemy session.
189+ The generated session can be used to access and query the database.
190+
191+ Returns
192+ -------
193+ None
194+
195+ Raises
196+ ------
197+ HTTPException
198+ If the converter list is not found or if there is an internal database error.
199+ """
200+ with session_factory () as db :
201+ try :
202+ converter_list = db .get (ConverterList , converter_list_id )
203+ if not converter_list :
204+ raise HTTPException (
205+ status_code = status .HTTP_404_NOT_FOUND ,
206+ detail = "Converter list not found" ,
207+ )
208+ notebook = converter_list .notebook
209+
210+ previous_converters = (
211+ db .query (ConverterList )
212+ .filter (
213+ ConverterList .notebook_id == converter_list .notebook_id ,
214+ ConverterList .created < converter_list .created ,
215+ )
216+ .all ()
217+ )
218+
219+ next_converters = (
220+ db .query (ConverterList )
221+ .filter (
222+ ConverterList .notebook_id == converter_list .notebook_id ,
223+ ConverterList .created >= converter_list .created ,
224+ )
225+ .all ()
226+ )
227+
228+ next_explorers = (
229+ db .query (Explorer )
230+ .filter (
231+ Explorer .notebook_id == converter_list .notebook_id ,
232+ Explorer .created >= converter_list .created ,
233+ )
234+ .all ()
235+ )
236+
237+ # Replace dataset from notebook with the original dataset
238+ shutil .copytree (
239+ notebook .dataset .file_path ,
240+ notebook .file_path ,
241+ dirs_exist_ok = True ,
242+ )
243+
244+ # Enqueue all previous converters
245+ for converter in previous_converters :
246+ await _enqueue_job_logic (
247+ job_type = "ConverterListJob" ,
248+ kwargs = {
249+ "converter_list_id" : converter .id ,
250+ },
251+ session_factory = session_factory ,
252+ component_registry = component_registry ,
253+ job_queue = job_queue ,
254+ )
255+
256+ app = request .app
257+ # Start the loop only if it's not already running or was cancelled
258+ if not hasattr (app .state , "job_loop" ) or app .state .job_loop .done ():
259+ app .state .job_loop = asyncio .create_task (job_queue_loop (True ))
260+
261+ # Delete all the converters after the current one
262+ for converter in next_converters :
263+ db .delete (converter )
264+
265+ # Delete all the explorers after the current converter
266+ for explorer in next_explorers :
267+ db .delete (explorer )
268+
269+ # Delete the current converter
270+ db .delete (converter_list )
271+ db .commit ()
272+
273+ except exc .SQLAlchemyError as e :
274+ logger .exception (e )
275+ raise HTTPException (
276+ status_code = status .HTTP_500_INTERNAL_SERVER_ERROR ,
277+ detail = "Internal database error" ,
278+ ) from e
0 commit comments