Mirrored_Repos/InvokeAI
1
0
Fork 0
mirror of https://github.com/invoke-ai/InvokeAI synced 2024-08-30 20:32:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

make session_list api return a raw dict rather than pydantic object

Browse Source
This commit is contained in:
Lincoln Stein 2023-08-17 22:22:06 -04:00
parent 832335998f
commit 86c11f9e27
2 changed files with 9 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
invokeai/app/api/routers/sessions.py
View File

@ -40,7 +40,7 @@ async def create_session(
@session_router.get(
"/",
operation_id="list_sessions",
responses={200: {"model": PaginatedResults[GraphExecutionState]}},
responses={200: {"model": PaginatedResults[dict]}},
)
async def list_sessions(
page: int = Query(default=0, description="The page of results to get"),

15
invokeai/app/services/sqlite.py
View File

@ -1,6 +1,7 @@
import sqlite3
import json
from threading import Lock
from typing import Generic, Optional, TypeVar, get_args
from typing import Generic, Optional, TypeVar, Union, get_args
from pydantic import BaseModel, parse_raw_as
@ -99,7 +100,7 @@ class SqliteItemStorage(ItemStorageABC, Generic[T]):
self._lock.release()
self._on_deleted(id)
def list(self, page: int = 0, per_page: int = 10) -> PaginatedResults[T]:
def list(self, page: int = 0, per_page: int = 10) -> PaginatedResults[dict]:
try:
self._lock.acquire()
self._cursor.execute(
@ -108,7 +109,7 @@ class SqliteItemStorage(ItemStorageABC, Generic[T]):
)
result = self._cursor.fetchall()
items = list(map(lambda r: self._parse_item(r[0]), result))
items = [json.loads(r[0]) for r in result]
self._cursor.execute(f"""SELECT count(*) FROM {self._table_name};""")
count = self._cursor.fetchone()[0]
@ -117,9 +118,9 @@ class SqliteItemStorage(ItemStorageABC, Generic[T]):
pageCount = int(count / per_page) + 1
return PaginatedResults[T](items=items, page=page, pages=pageCount, per_page=per_page, total=count)
return PaginatedResults[dict](items=items, page=page, pages=pageCount, per_page=per_page, total=count)
def search(self, query: str, page: int = 0, per_page: int = 10) -> PaginatedResults[T]:
def search(self, query: str, page: int = 0, per_page: int = 10) -> PaginatedResults[dict]:
try:
self._lock.acquire()
self._cursor.execute(
@ -128,7 +129,7 @@ class SqliteItemStorage(ItemStorageABC, Generic[T]):
)
result = self._cursor.fetchall()
items = list(map(lambda r: self._parse_item(r[0]), result))
items = [json.loads(r[0]) for r in result]
self._cursor.execute(
f"""SELECT count(*) FROM {self._table_name} WHERE item LIKE ?;""",
@ -140,4 +141,4 @@ class SqliteItemStorage(ItemStorageABC, Generic[T]):
pageCount = int(count / per_page) + 1
return PaginatedResults[T](items=items, page=page, pages=pageCount, per_page=per_page, total=count)
return PaginatedResults[dict](items=items, page=page, pages=pageCount, per_page=per_page, total=count)