Implementation Decisions¶

PgJobStore(
    dsn="postgresql://...",  # 옵션 A
    pool=existing_pool       # 옵션 B
)

async def handler(ctx: JobContext) -> None:
    # result를 어떻게 저장?

async def handler(ctx: JobContext) -> dict | None:
    return {"summary": "...", "tokens": 123}

# Worker가 자동으로 jobs.result에 저장

async def handler(ctx: JobContext) -> None:
    result = {"summary": "..."}
    # 옵션 B-1: ctx.set_result(result)
    # 옵션 B-2: 직접 DB 저장 (store.mark_done에서)

async def handler(ctx: JobContext) -> None:
    async with ctx.store.transaction() as conn:
        await conn.execute("INSERT INTO summaries ...")
    # jobs.result는 NULL

@dataclass(frozen=True)
class JobContext:
    job: Job
    store: PgJobStore
    worker_id: str
    conn: asyncpg.Connection  # ← 추가

async def handler(ctx: JobContext):
    await ctx.conn.execute("UPDATE ...")

async def handler(ctx: JobContext):
    async with ctx.store.transaction() as conn:
        await conn.execute("UPDATE ...")

import logging
logger = logging.getLogger("pqrun")

# Worker에서:
logger.info(f"Picked job {job.id} type={job.job_type}")

@dataclass(frozen=True)
class JobContext:
    job: Job
    store: PgJobStore
    worker_id: str
    logger: logging.Logger  # ← 추가

async def handler(ctx: JobContext):
    ctx.logger.info("Processing...")

# Handler에서 직접:
import logging
logger = logging.getLogger(__name__)

# lifespan finally:
stop_event.set()  # 새 job pickup 중지
await asyncio.gather(*tasks)  # 현재 job 완료까지 대기

for t in tasks:
    t.cancel()

stop_event.set()
# 1) 먼저 현재 실행 중인 job이 끝날 시간을 조금 준 뒤
await asyncio.wait_for(asyncio.gather(*worker_tasks, return_exceptions=True), timeout=shutdown_grace_s)
# 2) 그래도 안 끝나면 cancel로 강제 종료하고
for t in remaining_tasks:
    t.cancel()
# 3) 전체 shutdown은 bounded timeout 내에서 마무리
await asyncio.wait_for(asyncio.gather(*remaining_tasks, return_exceptions=True), timeout=shutdown_timeout_s)

await store.enqueue(job_type="x", dedupe_key="k1")
await store.enqueue(job_type="x", dedupe_key="k1")  # ← 중복

job_id = await store.enqueue(...)
if job_id == 0:
    # 이미 존재

raise DuplicateJobError("Job with dedupe_key already exists")

# Built-in handler
"__cleanup": cleanup_handler

# Auto-enqueued by Worker or pg_cron
DELETE FROM jobs
WHERE status IN ('DONE','FAILED','CANCELLED')
  AND finished_at < now() - interval '7 days';

import traceback
last_error = traceback.format_exc()

last_error = json.dumps({
    "type": type(e).__name__,
    "message": str(e),
    "traceback": traceback.format_exc()
})