feat(lotto-agent): get_agent_tasks 필터 + get_tasks_by_agent_date_kind 멱등 guard

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-23 01:52:05 +09:00
parent 0f65aa53e4
commit 4b6996b0f7

View File

@@ -236,12 +236,24 @@ def get_task(task_id: str) -> Optional[Dict[str, Any]]:
return _task_to_dict(r) if r else None
def get_agent_tasks(agent_id: str, limit: int = 20) -> List[Dict[str, Any]]:
def get_agent_tasks(
agent_id: str,
limit: int = 20,
task_type: Optional[str] = None,
days: Optional[int] = None,
) -> List[Dict[str, Any]]:
sql = "SELECT * FROM agent_tasks WHERE agent_id=?"
params: List[Any] = [agent_id]
if task_type is not None:
sql += " AND task_type=?"
params.append(task_type)
if days is not None and days > 0:
sql += " AND created_at >= datetime('now', ?)"
params.append(f"-{int(days)} days")
sql += " ORDER BY created_at DESC LIMIT ?"
params.append(limit)
with _conn() as conn:
rows = conn.execute(
"SELECT * FROM agent_tasks WHERE agent_id=? ORDER BY created_at DESC LIMIT ?",
(agent_id, limit),
).fetchall()
rows = conn.execute(sql, params).fetchall()
return [_task_to_dict(r) for r in rows]
@@ -739,3 +751,18 @@ def get_all_baselines() -> List[Dict[str, Any]]:
d["window_values"] = json.loads(d["window_values"])
out.append(d)
return out
def get_tasks_by_agent_date_kind(agent_id: str, date_iso: str, task_type: str) -> List[Dict[str, Any]]:
"""같은 (agent, date, task_type)으로 이미 생성된 task 조회. 멱등 guard."""
with _conn() as conn:
rows = conn.execute(
"""
SELECT * FROM agent_tasks
WHERE agent_id = ? AND task_type = ?
AND substr(created_at, 1, 10) = ?
ORDER BY created_at DESC
""",
(agent_id, task_type, date_iso),
).fetchall()
return [_task_to_dict(r) for r in rows]