mirror of
https://git.datalinker.icu/comfyanonymous/ComfyUI
synced 2026-09-02 04:47:05 +08:00
make execution.py more pythonic
This commit is contained in:
parent
80ea3d81af
commit
4a2172758a
25
execution.py
25
execution.py
@ -1116,22 +1116,25 @@ class PromptQueue:
|
||||
else:
|
||||
return {}
|
||||
|
||||
def get_ordered_history(self, max_items=None, offset=-1):
|
||||
def get_ordered_history(self, max_items=None, offset=0):
|
||||
with self.mutex:
|
||||
out = []
|
||||
history_keys = list(self.history.keys())
|
||||
|
||||
i = 0
|
||||
if offset < 0 and max_items is not None:
|
||||
offset = len(self.history) - max_items
|
||||
for k in self.history:
|
||||
if i >= offset:
|
||||
out.append({k: self.history[k]})
|
||||
if max_items is not None and len(out) >= max_items:
|
||||
break
|
||||
i += 1
|
||||
offset = max(0, len(history_keys) - max_items)
|
||||
|
||||
# Use slice to get the desired range
|
||||
end_index = offset + max_items if max_items is not None else None
|
||||
selected_keys = history_keys[offset:end_index]
|
||||
|
||||
return {"history": out}
|
||||
# Build history items with prompt_id field
|
||||
history_items = []
|
||||
for key in selected_keys:
|
||||
item = copy.deepcopy(self.history[key])
|
||||
item["prompt_id"] = key
|
||||
history_items.append(item)
|
||||
|
||||
return {"history": history_items}
|
||||
|
||||
def wipe_history(self):
|
||||
with self.mutex:
|
||||
|
||||
@ -658,11 +658,8 @@ class PromptServer():
|
||||
if max_items is not None:
|
||||
max_items = int(max_items)
|
||||
|
||||
offset = request.rel_url.query.get("offset", None)
|
||||
if offset is not None:
|
||||
offset = int(offset)
|
||||
else:
|
||||
offset = -1
|
||||
offset = request.rel_url.query.get("offset", 0)
|
||||
offset = int(offset)
|
||||
|
||||
return web.json_response(self.prompt_queue.get_ordered_history(max_items=max_items, offset=offset))
|
||||
|
||||
|
||||
@ -638,10 +638,7 @@ class TestExecution:
|
||||
assert len(ordered_history["history"]) == 3, "Should have exactly 3 prompts in history"
|
||||
|
||||
# Verify chronological ordering (most recent first)
|
||||
history_prompt_ids = []
|
||||
for item in ordered_history["history"]:
|
||||
for prompt_id in item.keys():
|
||||
history_prompt_ids.append(prompt_id)
|
||||
history_prompt_ids = [item["prompt_id"] for item in ordered_history["history"]]
|
||||
|
||||
# Should be in chronological order (oldest first, as they're added in completion order)
|
||||
assert history_prompt_ids == prompt_ids, "History should be in chronological order"
|
||||
@ -707,38 +704,26 @@ class TestExecution:
|
||||
assert len(full_history["history"]) == 5, "Should have 5 items in full history"
|
||||
|
||||
# Extract prompt IDs from full history for comparison
|
||||
full_prompt_ids = []
|
||||
for item in full_history["history"]:
|
||||
for prompt_id in item.keys():
|
||||
full_prompt_ids.append(prompt_id)
|
||||
full_prompt_ids = [item["prompt_id"] for item in full_history["history"]]
|
||||
|
||||
# Test proper pagination behavior with offset as cursor
|
||||
|
||||
# Test first page: offset=0, max_items=2
|
||||
page1 = client.get_ordered_history(max_items=2, offset=0)
|
||||
assert len(page1["history"]) == 2, "First page should have 2 items"
|
||||
page1_ids = []
|
||||
for item in page1["history"]:
|
||||
for prompt_id in item.keys():
|
||||
page1_ids.append(prompt_id)
|
||||
page1_ids = [item["prompt_id"] for item in page1["history"]]
|
||||
assert page1_ids == full_prompt_ids[0:2], "First page should contain items at indices 0-1"
|
||||
|
||||
# Test second page: offset=2, max_items=2
|
||||
page2 = client.get_ordered_history(max_items=2, offset=2)
|
||||
assert len(page2["history"]) == 2, "Second page should have 2 items"
|
||||
page2_ids = []
|
||||
for item in page2["history"]:
|
||||
for prompt_id in item.keys():
|
||||
page2_ids.append(prompt_id)
|
||||
page2_ids = [item["prompt_id"] for item in page2["history"]]
|
||||
assert page2_ids == full_prompt_ids[2:4], "Second page should contain items at indices 2-3"
|
||||
|
||||
# Test third page: offset=4, max_items=2
|
||||
page3 = client.get_ordered_history(max_items=2, offset=4)
|
||||
assert len(page3["history"]) == 1, "Third page should have 1 remaining item"
|
||||
page3_ids = []
|
||||
for item in page3["history"]:
|
||||
for prompt_id in item.keys():
|
||||
page3_ids.append(prompt_id)
|
||||
page3_ids = [item["prompt_id"] for item in page3["history"]]
|
||||
assert page3_ids == full_prompt_ids[4:5], "Third page should contain item at index 4"
|
||||
|
||||
# Verify no overlap between pages
|
||||
@ -750,10 +735,7 @@ class TestExecution:
|
||||
# When offset < 0 and max_items is specified, offset = len(history) - max_items
|
||||
last_2_items = client.get_ordered_history(max_items=2)
|
||||
assert len(last_2_items["history"]) == 2, "Default behavior should return 2 items"
|
||||
last_2_ids = []
|
||||
for item in last_2_items["history"]:
|
||||
for prompt_id in item.keys():
|
||||
last_2_ids.append(prompt_id)
|
||||
last_2_ids = [item["prompt_id"] for item in last_2_items["history"]]
|
||||
# This should be equivalent to offset=3 (5-2=3)
|
||||
assert last_2_ids == full_prompt_ids[3:5], "Default behavior should return last 2 items"
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user