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

feat(batch): extract repeated logic to function

Browse Source
This commit is contained in:
psychedelicious 2023-09-05 16:14:18 +10:00
parent e9b26e5e7d
commit babd26feab
1 changed files with 10 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
invokeai/app/services/batch_manager.py
View File

@ -178,21 +178,19 @@ class BatchManager(BatchManagerBase):
def get_batch_processes(self) -> list[BatchProcessResponse]: def get_batch_processes(self) -> list[BatchProcessResponse]:
bps = self.__batch_process_storage.get_all() bps = self.__batch_process_storage.get_all()
res = list() return self._get_batch_process_responses(bps)
for bp in bps:
sessions = self.__batch_process_storage.get_sessions_by_batch_id(bp.batch_id)
res.append(
BatchProcessResponse(
batch_id=bp.batch_id,
session_ids=[session.session_id for session in sessions],
)
)
return res
def get_incomplete_batch_processes(self) -> list[BatchProcessResponse]: def get_incomplete_batch_processes(self) -> list[BatchProcessResponse]:
bps = self.__batch_process_storage.get_incomplete() bps = self.__batch_process_storage.get_incomplete()
res = list() return self._get_batch_process_responses(bps)
for bp in bps:
def cancel_batch_process(self, batch_process_id: str) -> None:
self.__batch_process_storage.cancel(batch_process_id)
def _get_batch_process_responses(self, batch_processes: list[BatchProcess]) -> list[BatchProcessResponse]:
sessions = list()
res: list[BatchProcessResponse] = list()
for bp in batch_processes:
sessions = self.__batch_process_storage.get_sessions_by_batch_id(bp.batch_id) sessions = self.__batch_process_storage.get_sessions_by_batch_id(bp.batch_id)
res.append( res.append(
BatchProcessResponse( BatchProcessResponse(
@ -201,6 +199,3 @@ class BatchManager(BatchManagerBase):
) )
) )
return res return res
def cancel_batch_process(self, batch_process_id: str) -> None:
self.__batch_process_storage.cancel(batch_process_id)