@@ -24,15 +24,15 @@ class SqliteWorkflowRecordsStorage(WorkflowRecordsStorageBase):
24
24
def __init__ (self , db : SqliteDatabase ) -> None :
25
25
super ().__init__ ()
26
26
self ._conn = db .conn
27
- self ._cursor = self ._conn .cursor ()
28
27
29
28
def start (self , invoker : Invoker ) -> None :
30
29
self ._invoker = invoker
31
30
self ._sync_default_workflows ()
32
31
33
32
def get (self , workflow_id : str ) -> WorkflowRecordDTO :
34
33
"""Gets a workflow by ID. Updates the opened_at column."""
35
- self ._cursor .execute (
34
+ cursor = self ._conn .cursor ()
35
+ cursor .execute (
36
36
"""--sql
37
37
UPDATE workflow_library
38
38
SET opened_at = STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')
@@ -41,15 +41,15 @@ def get(self, workflow_id: str) -> WorkflowRecordDTO:
41
41
(workflow_id ,),
42
42
)
43
43
self ._conn .commit ()
44
- self . _cursor .execute (
44
+ cursor .execute (
45
45
"""--sql
46
46
SELECT workflow_id, workflow, name, created_at, updated_at, opened_at
47
47
FROM workflow_library
48
48
WHERE workflow_id = ?;
49
49
""" ,
50
50
(workflow_id ,),
51
51
)
52
- row = self . _cursor .fetchone ()
52
+ row = cursor .fetchone ()
53
53
if row is None :
54
54
raise WorkflowNotFoundError (f"Workflow with id { workflow_id } not found" )
55
55
return WorkflowRecordDTO .from_dict (dict (row ))
@@ -59,7 +59,8 @@ def create(self, workflow: WorkflowWithoutID) -> WorkflowRecordDTO:
59
59
# Only user workflows may be created by this method
60
60
assert workflow .meta .category is WorkflowCategory .User
61
61
workflow_with_id = Workflow (** workflow .model_dump (), id = uuid_string ())
62
- self ._cursor .execute (
62
+ cursor = self ._conn .cursor ()
63
+ cursor .execute (
63
64
"""--sql
64
65
INSERT OR IGNORE INTO workflow_library (
65
66
workflow_id,
@@ -77,7 +78,8 @@ def create(self, workflow: WorkflowWithoutID) -> WorkflowRecordDTO:
77
78
78
79
def update (self , workflow : Workflow ) -> WorkflowRecordDTO :
79
80
try :
80
- self ._cursor .execute (
81
+ cursor = self ._conn .cursor ()
82
+ cursor .execute (
81
83
"""--sql
82
84
UPDATE workflow_library
83
85
SET workflow = ?
@@ -93,7 +95,8 @@ def update(self, workflow: Workflow) -> WorkflowRecordDTO:
93
95
94
96
def delete (self , workflow_id : str ) -> None :
95
97
try :
96
- self ._cursor .execute (
98
+ cursor = self ._conn .cursor ()
99
+ cursor .execute (
97
100
"""--sql
98
101
DELETE from workflow_library
99
102
WHERE workflow_id = ? AND category = 'user';
@@ -149,12 +152,13 @@ def get_many(
149
152
main_query += " LIMIT ? OFFSET ?"
150
153
main_params .extend ([per_page , page * per_page ])
151
154
152
- self ._cursor .execute (main_query , main_params )
153
- rows = self ._cursor .fetchall ()
155
+ cursor = self ._conn .cursor ()
156
+ cursor .execute (main_query , main_params )
157
+ rows = cursor .fetchall ()
154
158
workflows = [WorkflowRecordListItemDTOValidator .validate_python (dict (row )) for row in rows ]
155
159
156
- self . _cursor .execute (count_query , count_params )
157
- total = self . _cursor .fetchone ()[0 ]
160
+ cursor .execute (count_query , count_params )
161
+ total = cursor .fetchone ()[0 ]
158
162
159
163
if per_page :
160
164
pages = total // per_page + (total % per_page > 0 )
@@ -193,14 +197,15 @@ def _sync_default_workflows(self) -> None:
193
197
workflows .append (workflow )
194
198
# Only default workflows may be managed by this method
195
199
assert all (w .meta .category is WorkflowCategory .Default for w in workflows )
196
- self ._cursor .execute (
200
+ cursor = self ._conn .cursor ()
201
+ cursor .execute (
197
202
"""--sql
198
203
DELETE FROM workflow_library
199
204
WHERE category = 'default';
200
205
"""
201
206
)
202
207
for w in workflows :
203
- self . _cursor .execute (
208
+ cursor .execute (
204
209
"""--sql
205
210
INSERT OR REPLACE INTO workflow_library (
206
211
workflow_id,
0 commit comments