31 lines
805 B
Python
31 lines
805 B
Python
import json
|
|
from typing import Any
|
|
|
|
import asyncpg
|
|
|
|
JOBS_CHANGED_CHANNEL = "jobs_changed"
|
|
|
|
|
|
async def enqueue(
|
|
pool: asyncpg.Pool, account_id: int, kind: str, params: dict[str, Any]
|
|
) -> int:
|
|
params_json = json.dumps(params)
|
|
existing = await pool.fetchval(
|
|
"SELECT id FROM jobs WHERE account_id = $1 AND kind = $2 "
|
|
"AND params = $3::jsonb AND status IN ('pending', 'running') LIMIT 1",
|
|
account_id,
|
|
kind,
|
|
params_json,
|
|
)
|
|
if existing is not None:
|
|
return existing
|
|
job_id = await pool.fetchval(
|
|
"INSERT INTO jobs (account_id, kind, params) "
|
|
"VALUES ($1, $2, $3::jsonb) RETURNING id",
|
|
account_id,
|
|
kind,
|
|
params_json,
|
|
)
|
|
await pool.execute(f"NOTIFY {JOBS_CHANGED_CHANNEL}")
|
|
return job_id
|