### Complete Producer-Consumer Job Queue Workflow

Source: https://context7.com/adriangb/pgjobq/llms.txt

This example shows a full job queue implementation. It sets up a PostgreSQL pool, migrates the schema, creates a queue, and then uses multiple workers to consume jobs while a producer sends jobs and waits for their completion. Ensure you have PostgreSQL running and the 'pgjobq' library installed.

```python
import asyncpg
import anyio
from contextlib import AsyncExitStack
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version

async def main():
    async with AsyncExitStack() as stack:
        pool = await stack.enter_async_context(
            asyncpg.create_pool("postgres://postgres:postgres@localhost/postgres")
        )
        await migrate_to_latest_version(pool)
        await create_queue("myqueue", pool)

        queue = await stack.enter_async_context(
            connect_to_queue("myqueue", pool)
        )

        async with anyio.create_task_group() as tg:
            async def worker():
                async with queue.receive() as job_stream:
                    async with (await job_stream.receive()).acquire() as job:
                        print("received")
                        await anyio.sleep(1)  # Simulate work
                        print("done processing")
                    print("acked")

            tg.start_soon(worker)
            tg.start_soon(worker)  # Multiple workers

            async with queue.send(b'{"foo":"bar"}') as completion_handle:
                print("sent")
                await completion_handle.wait()
                print("completed")
                tg.cancel_scope.cancel()

if __name__ == "__main__":
    anyio.run(main)
    # Output:
    # sent
    # received
    # done processing
    # acked
    # completed
```

--------------------------------

### Create and Use a pgjobq Queue

Source: https://github.com/adriangb/pgjobq/blob/main/README.md

This example demonstrates setting up a PostgreSQL connection, migrating the queue schema, creating a queue, and then using it to send and receive jobs concurrently with multiple workers. It utilizes asyncpg for database interaction and anyio for asynchronous task management.

```python
from contextlib import AsyncExitStack

import anyio
import asyncpg  # type: ignore
from pgjobq import create_queue, connect_to_queue, migrate_to_latest_version

async def main() -> None:

    async with AsyncExitStack() as stack:
        pool: asyncpg.Pool = await stack.enter_async_context(
            asyncpg.create_pool(  # type: ignore
                "postgres://postgres:postgres@localhost/postgres"
            )
        )
        await migrate_to_latest_version(pool)
        await create_queue("myq", pool)
        queue = await stack.enter_async_context(
            connect_to_queue("myq", pool)
        )
        async with anyio.create_task_group() as tg:

            async def worker() -> None:
                async with queue.receive() as msg_handle_rcv_stream:
                    # receive a single job
                    async with (await msg_handle_rcv_stream.receive()).acquire():
                        print("received")
                        # do some work
                        await anyio.sleep(1)
                        print("done processing")
                        print("acked")

            tg.start_soon(worker)
            tg.start_soon(worker)

            async with queue.send(b'{"foo":"bar"}') as completion_handle:
                print("sent")
                await completion_handle.wait()
                print("completed")
                tg.cancel_scope.cancel()


if __name__ == "__main__":
    anyio.run(main)
    # prints:
    # "sent"
    # "received"
    # "done processing"
    # "acked"
    # "completed"

```

--------------------------------

### Handle Backpressure with Bounded Queues

Source: https://context7.com/adriangb/pgjobq/llms.txt

Implement a producer-consumer pattern where the producer waits if the queue reaches its maximum size. This example uses `anyio` for task management and `queue.wait_if_full()` to respect backpressure.

```python
import asyncpg
import anyio
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version

async def backpressure_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("bounded-queue", pool, max_size=5)

    async with connect_to_queue("bounded-queue", pool) as queue:
        # Worker that slowly processes jobs
        async def slow_worker():
            async with queue.receive() as job_stream:
                count = 0
                async for job_handle in job_stream:
                    async with job_handle.acquire() as job:
                        await anyio.sleep(0.5)  # Slow processing
                        count += 1
                    if count >= 10:
                        break

        # Producer that respects backpressure
        async def producer():
            for i in range(10):
                # Wait if queue is full
                await queue.wait_if_full()
                async with queue.send(f'{{"job": {i}}}'.encode()):
                    print(f"Sent job {i}")

        async with anyio.create_task_group() as tg:
            tg.start_soon(slow_worker)
            tg.start_soon(producer)

    await pool.close()
```

--------------------------------

### Process Dead Letter Queue (DLQ)

Source: https://context7.com/adriangb/pgjobq/llms.txt

Access and process jobs from the dead-letter queue, which holds jobs that failed delivery. This example demonstrates setting a low `max_delivery_attempts` to move jobs to the DLQ and then processing them.

```python
import asyncpg
from datetime import timedelta
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version, get_dlq_name

async def dlq_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)

    # Create queue with low retry limit
    await create_queue(
        "failing-queue",
        pool,
        max_delivery_attempts=2,  # Move to DLQ after 2 failures
    )

    # Simulate a failing worker
    async with connect_to_queue("failing-queue", pool) as queue:
        async with queue.send(b'{"will": "fail"}'):
            pass

        # Process and fail twice
        for attempt in range(2):
            try:
                async with queue.receive() as job_stream:
                    async for job_handle in job_stream:
                        async with job_handle.acquire() as job:
                            print(f"Attempt {attempt + 1}: Processing {job.id}")
                            raise ValueError("Simulated failure")
            except ValueError:
                print(f"Attempt {attempt + 1} failed")

    # Process DLQ
    dlq_name = get_dlq_name("failing-queue")  # Returns "dlq@failing-queue"

    async with connect_to_queue(dlq_name, pool) as dlq:
        async with dlq.receive() as job_stream:
            async for job_handle in job_stream:
                async with job_handle.acquire() as job:
                    print(f"DLQ job: {job.id}, body: {job.body}")
                    # Handle dead letter (log, alert, manual retry, etc.)
                break

    await pool.close()
```

--------------------------------

### Get Queue Statistics

Source: https://context7.com/adriangb/pgjobq/llms.txt

Retrieve current job count and max size configuration for a queue. Note that stats are cached and may require reconnecting for fresh data.

```python
import asyncpg
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version

async def stats_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("stats-queue", pool, max_size=100)

    async with connect_to_queue("stats-queue", pool) as queue:
        # Get initial stats
        stats = await queue.get_statistics()
        print(f"Jobs in queue: {stats.jobs}")
        print(f"Max queue size: {stats.max_size}")

        # Send some jobs
        async with queue.send(b'{"job": 1}', b'{"job": 2}'):
            pass

        # Stats are cached; reconnect for fresh stats
        stats = await queue.get_statistics()
        print(f"Jobs after sending: {stats.jobs}")

    await pool.close()
```

--------------------------------

### Create a New Queue with Custom Options

Source: https://context7.com/adriangb/pgjobq/llms.txt

Create a new queue in PostgreSQL with configurable options such as acknowledgment deadline, max delivery attempts, retention period, and backpressure limits. Ensures migrations are applied first.

```python
import asyncpg
from datetime import timedelta
from pgjobq import create_queue, migrate_to_latest_version

async def create_my_queue():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)

    # Create a queue with custom options
    created = await create_queue(
        "email-notifications",
        pool,
        ack_deadline=timedelta(seconds=30),
        max_delivery_attempts=5,
        retention_period=timedelta(days=7),
        max_size=1000,
        backoff_power_base=2,
    )

    if created:
        print("Queue created successfully")
    else:
        print("Queue already exists")

    await pool.close()
```

--------------------------------

### Concurrent Job Processing with PGJobQ and AnyIO

Source: https://context7.com/adriangb/pgjobq/llms.txt

This snippet demonstrates processing multiple jobs concurrently using AnyIO task groups for maximum throughput. It sets up a connection pool, migrates the queue, creates a queue, and then defines a function to process individual jobs concurrently.

```python
import asyncpg
import anyio
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version
from pgjobq.api import JobHandle

async def concurrent_worker():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("concurrent-queue", pool)

    async def process_single_job(job_handle: JobHandle) -> None:
        async with job_handle.acquire() as job:
            print(f"Processing {job.id}")
            await anyio.sleep(0.25)  # Simulate work
            print(f"Completed {job.id}")

    async with connect_to_queue("concurrent-queue", pool) as queue:
        total_jobs = 15

        # Receive and process jobs concurrently
        async with queue.receive(batch_size=total_jobs) as job_stream:
            async with anyio.create_task_group() as worker_tg:
                n = total_jobs
                async for job_handle in job_stream:
                    # Start each job in its own task
                    worker_tg.start_soon(process_single_job, job_handle)
                    n -= 1
                    if n == 0:
                        break

    await pool.close()
```

--------------------------------

### Run Database Migrations with pgjobq

Source: https://context7.com/adriangb/pgjobq/llms.txt

Apply all pending pgjobq schema migrations to your PostgreSQL database. Requires an asyncpg pool connection.

```python
import asyncpg
from pgjobq import migrate_to_latest_version

async def setup_database():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )

    # Apply all pending migrations
    await migrate_to_latest_version(pool)

    await pool.close()
```

--------------------------------

### Connect to an Existing Queue

Source: https://context7.com/adriangb/pgjobq/llms.txt

Establish a connection to an existing pgjobq queue. This connection manages background tasks for cleanup and heartbeats. Ensures migrations are applied and the queue exists before connecting.

```python
import asyncpg
from contextlib import AsyncExitStack
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version

async def connect_example():
    async with AsyncExitStack() as stack:
        pool = await stack.enter_async_context(
            asyncpg.create_pool("postgres://postgres:postgres@localhost/postgres")
        )
        await migrate_to_latest_version(pool)
        await create_queue("myqueue", pool)

        # Connect to the queue
        queue = await stack.enter_async_context(
            connect_to_queue("myqueue", pool)
        )

        # Queue is now ready for sending/receiving jobs
        print(f"Connected to queue")
```

--------------------------------

### Batch Job Receiving with PGJobQ

Source: https://context7.com/adriangb/pgjobq/llms.txt

Use this snippet to receive multiple jobs at once for improved throughput in high-volume scenarios. It configures a pool, migrates the queue, creates a queue, and then connects to it to receive jobs in batches.

```python
import asyncpg
import anyio
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version

async def batch_worker():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("batch-queue", pool)

    async with connect_to_queue("batch-queue", pool) as queue:
        # Receive up to 10 jobs at a time, poll every 2 seconds
        async with queue.receive(batch_size=10, poll_interval=2) as job_stream:
            jobs_processed = 0

            async for job_handle in job_stream:
                async with job_handle.acquire() as job:
                    print(f"Processing job {jobs_processed + 1}: {job.body}")
                    await anyio.sleep(0.01)  # Simulate fast processing

                jobs_processed += 1
                if jobs_processed >= 100:
                    break

            print(f"Processed {jobs_processed} jobs")

    await pool.close()
```

--------------------------------

### Create Job Dependencies

Source: https://context7.com/adriangb/pgjobq/llms.txt

Establish job dependencies to ensure ordered processing. The dependent job will only become available after the parent job completes successfully. Ensure the pool and queue are set up.

```python
import asyncpg
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version, OutgoingJob

async def job_dependencies_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("workflow", pool)

    async with connect_to_queue("workflow", pool) as queue:
        # First, send the job that must complete first
        async with queue.send(b'{"step": "download_data"}') as handle1:
            parent_job_ids = list(handle1.jobs.keys())
            print(f"Parent job ID: {parent_job_ids[0]}")

            # Send a dependent job that won't be available until parent completes
            dependent_job = OutgoingJob(
                body=b'{"step": "process_data"}',
                dependencies=parent_job_ids  # Must wait for parent
            )
            async with queue.send(dependent_job) as handle2:
                print("Dependent job sent")

                # Worker will receive jobs in dependency order:
                # process_data is delivered first (since download_data blocks it)
                # download_data becomes available after process_data completes

    await pool.close()
```

--------------------------------

### Wait for Job Completion

Source: https://context7.com/adriangb/pgjobq/llms.txt

Monitor and wait for specific jobs to complete without sending them. This is useful for coordination and monitoring tasks. Requires importing asyncpg, anyio, and pgjobq components.

```python
import asyncpg
import anyio
from datetime import timedelta
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version

async def wait_for_completion_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("completion-queue", pool)

    job_ids = []

    # Producer: send jobs and collect IDs
    async with connect_to_queue("completion-queue", pool) as queue:
        async with queue.send(b'{"job": 1}') as handle:
            job_ids.extend(handle.jobs.keys())
        async with queue.send(b'{"job": 2}') as handle:
            job_ids.extend(handle.jobs.keys())

    # Monitor: wait for jobs in a separate connection
    async def monitor():
        async with connect_to_queue("completion-queue", pool) as queue:
            async with queue.wait_for_completion(
                *job_ids,
                poll_interval=timedelta(seconds=1)
            ) as handle:
                print("Waiting for jobs to complete...")
                await handle.wait()
                print("All jobs completed!")

    # Worker: process jobs
    async def worker():
        async with connect_to_queue("completion-queue", pool) as queue:
            async with queue.receive() as job_stream:
                count = 0
                async for job_handle in job_stream:
                    async with job_handle.acquire() as job:
                        print(f"Processing {job.id}")
                        await anyio.sleep(0.1)
                    count += 1
                    if count >= 2:
                        break

    async with anyio.create_task_group() as tg:
        tg.start_soon(monitor)
        tg.start_soon(worker)

    await pool.close()
```

--------------------------------

### Implement Custom Logging Telemetry Hook

Source: https://context7.com/adriangb/pgjobq/llms.txt

Create a custom telemetry hook by subclassing TelemetryHook and implementing the on_query method to log query details. This hook can be applied at the connection level or overridden for specific operations.

```python
import asyncpg
from contextlib import asynccontextmanager
from typing import Any, AsyncIterator, Sequence
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version, TelemetryHook
from pgjobq.types import PoolOrConnection

class LoggingTelemetryHook:
    """Custom telemetry hook that logs all queries"""

    @asynccontextmanager
    async def on_query(
        self,
        queue_name: str,
        operation: str,
        query: str,
        conn: PoolOrConnection,
        args: Sequence[Any],
    ) -> AsyncIterator[PoolOrConnection]:
        print(f"[{queue_name}] {operation}")
        print(f"  Query: {query[:100]}...")
        print(f"  Args: {args}")
        yield conn
        print(f"[{queue_name}] {operation} completed")

class ExplainTelemetryHook:
    """Hook that runs EXPLAIN ANALYZE on queries"""

    @asynccontextmanager
    async def on_query(
        self,
        queue_name: str,
        operation: str,
        query: str,
        conn: PoolOrConnection,
        args: Sequence[Any],
    ) -> AsyncIterator[PoolOrConnection]:
        # Run EXPLAIN before actual query
        if operation in ("poll", "publish"):
            explain_query = f"EXPLAIN ANALYZE {query}"
            # result = await conn.fetch(explain_query, *args)
            # print(f"Query plan for {operation}:", result)
        yield conn

async def telemetry_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("monitored-queue", pool)

    hook = LoggingTelemetryHook()

    # Use hook at connection level
    async with connect_to_queue("monitored-queue", pool, telemetry_hook=hook) as queue:
        async with queue.send(b'{"monitored": true}'):
            pass

        # Or override per-operation
        async with queue.receive(telemetry_hook=hook) as job_stream:
            async for job_handle in job_stream:
                async with job_handle.acquire():
                    pass
                break

    await pool.close()
```

--------------------------------

### Send Jobs to a Queue

Source: https://context7.com/adriangb/pgjobq/llms.txt

Send one or multiple jobs to a pgjobq queue. Supports sending raw bytes, batching, waiting for job completion, and scheduling jobs for a future time. Requires connecting to the queue first.

```python
import asyncpg
from datetime import datetime, timedelta
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version

async def send_jobs_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("tasks", pool)

    async with connect_to_queue("tasks", pool) as queue:
        # Send a single job (bytes)
        async with queue.send(b'{"action": "send_email", "to": "user@example.com"}'):
            print("Job sent")

        # Send multiple jobs in batch
        async with queue.send(b'{"task": 1}', b'{"task": 2}', b'{"task": 3}'):
            print("Batch sent")

        # Send and wait for completion
        async with queue.send(b'{"important": true}') as completion_handle:
            print("Job sent, waiting for worker to complete...")
            await completion_handle.wait()
            print("Job completed!")

        # Schedule a job for later
        scheduled_time = datetime.utcnow() + timedelta(minutes=5)
        async with queue.send(b'{"delayed": true}', schedule_at=scheduled_time):
            print("Job scheduled for 5 minutes from now")

    await pool.close()
```

--------------------------------

### Receive and Process Jobs

Source: https://context7.com/adriangb/pgjobq/llms.txt

Implement a worker to receive jobs from a queue. Jobs are automatically acknowledged on success and redelivered on failure. Ensure the pool and queue are set up.

```python
import asyncpg
import anyio
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version

async def worker_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("tasks", pool)

    async with connect_to_queue("tasks", pool) as queue:
        # Basic job receiving pattern
        async with queue.receive() as job_handle_stream:
            async for job_handle in job_handle_stream:
                # acquire() provides the job and handles ack/nack
                async with job_handle.acquire() as job:
                    print(f"Processing job {job.id}")
                    print(f"Body: {job.body}")
                    print(f"Attributes: {job.attributes}")

                    # Do your work here
                    await process_job(job.body)

                    # Job is automatically acknowledged when context exits successfully
                    # If an exception is raised, job is nacked and redelivered

                # Process only one job then exit
                break

    await pool.close()

async def process_job(body: bytes):
    # Simulate work
    await anyio.sleep(0.1)
    print(f"Processed: {body.decode()}")
```

--------------------------------

### Filter Jobs by Attributes in Python

Source: https://context7.com/adriangb/pgjobq/llms.txt

Use attribute filters to receive only jobs matching specific criteria. Supports comparison operators, null checks, pattern matching, and logical combinations. Ensure the asyncpg pool is properly configured and the queue is migrated.

```python
import asyncpg
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version, Attribute, OutgoingJob

async def filtered_worker():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("filtered-queue", pool)

    async with connect_to_queue("filtered-queue", pool) as queue:
        # Send jobs with different attributes
        async with queue.send(OutgoingJob(b'{"low": true}', {"priority": 1})):
            pass
        async with queue.send(OutgoingJob(b'{"high": true}', {"priority": 10})):
            pass
        async with queue.send(OutgoingJob(b'{"urgent": true}', {"priority": 100, "region": "us-east"})):
            pass

        # Filter: priority > 5
        high_priority = Attribute("priority").gt(5)

        # Filter: region equals "us-east"
        us_east = Attribute("region").eq("us-east")

        # Combine filters with AND/OR
        urgent_or_high = high_priority | us_east

        # Receive only matching jobs
        async with queue.receive(filter=high_priority) as job_stream:
            async for job_handle in job_stream:
                async with job_handle.acquire() as job:
                    print(f"High priority job: {job.body}, attrs: {job.attributes}")
                break

        # Other filter operations
        filters = [
            Attribute("status").eq("pending"),          # Equality
            Attribute("status").ne("completed"),        # Not equal
            Attribute("count").lt(10),                  # Less than
            Attribute("count").le(10),                  # Less than or equal
            Attribute("count").gt(5),                   # Greater than
            Attribute("count").ge(5),                   # Greater than or equal
            Attribute("name").is_like("user%"),         # SQL LIKE pattern
            Attribute("name").is_not_like("%test%"),    # SQL NOT LIKE
            Attribute("value").is_null(),               # IS NULL
            Attribute("value").is_not_null(),           # IS NOT NULL
            Attribute("optional").exists(),             # Attribute exists
            Attribute("optional").does_not_exist(),     # Attribute doesn't exist
        ]

        # Complex filter with AND and OR
        complex_filter = (
            (Attribute("priority").gt(5) & Attribute("region").eq("us-west"))
            | Attribute("urgent").eq(True)
        )

    await pool.close()
```

```python
Attribute("priority").gt(5)
```

```python
Attribute("region").eq("us-east")
```

```python
high_priority | us_east
```

```python
Attribute("status").eq("pending")
```

```python
Attribute("status").ne("completed")
```

```python
Attribute("count").lt(10)
```

```python
Attribute("count").le(10)
```

```python
Attribute("count").gt(5)
```

```python
Attribute("count").ge(5)
```

```python
Attribute("name").is_like("user%")
```

```python
Attribute("name").is_not_like("%test%")
```

```python
Attribute("value").is_null()
```

```python
Attribute("value").is_not_null()
```

```python
Attribute("optional").exists()
```

```python
Attribute("optional").does_not_exist()
```

```python
(Attribute("priority").gt(5) & Attribute("region").eq("us-west")) | Attribute("urgent").eq(True)
```

--------------------------------

### Send Jobs with Attributes using OutgoingJob

Source: https://context7.com/adriangb/pgjobq/llms.txt

Send jobs with custom attributes for filtering. Attributes can be strings, numbers, or booleans. Ensure the pool and queue are set up before sending.

```python
import asyncpg
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version, OutgoingJob

async def send_with_attributes():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("notifications", pool)

    async with connect_to_queue("notifications", pool) as queue:
        # Send job with attributes using OutgoingJob
        job = OutgoingJob(
            body=b'{"message": "Hello!"}',
            attributes={
                "priority": "high",
                "user_id": 12345,
                "region": "us-west",
                "retry_count": 0,
            }
        )
        async with queue.send(job):
            print("Job with attributes sent")

        # Send multiple jobs with different attributes
        jobs = [
            OutgoingJob(b'{"type": "email"}', {"channel": "email", "priority": 1}),
            OutgoingJob(b'{"type": "sms"}', {"channel": "sms", "priority": 2}),
            OutgoingJob(b'{"type": "push"}', {"channel": "push", "priority": 3}),
        ]
        async with queue.send(*jobs):
            print("Multiple attributed jobs sent")

    await pool.close()
```

--------------------------------

### Handle QueueDoesNotExist Exception

Source: https://context7.com/adriangb/pgjobq/llms.txt

Catch the QueueDoesNotExist exception when attempting to send messages to a queue that has not been created. Ensure queues exist before operations that require them.

```python
import asyncpg
from pgjobq import (
    connect_to_queue,
    create_queue,
    migrate_to_latest_version,
    JobDoesNotExist,
    QueueDoesNotExist,
    ReceiptHandleExpired,
)

async def exception_handling_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)

    # QueueDoesNotExist - raised when sending to non-existent queue
    async with connect_to_queue("nonexistent-queue", pool) as queue:
        try:
            async with queue.send(b'{"test": true}'):
                pass
        except QueueDoesNotExist as e:
            print(f"Queue not found: {e.queue_name}")

    # Create queue for further examples
    await create_queue("error-demo", pool)

    async with connect_to_queue("error-demo", pool) as queue:
        # ReceiptHandleExpired - job was redelivered to another worker
        # This typically happens when processing takes longer than ack_deadline
        try:
            async with queue.send(b'{"slow": true}'):
                pass
            async with queue.receive() as job_stream:
                async for job_handle in job_stream:
                    async with job_handle.acquire() as job:
                        # If this takes too long, receipt handle may expire
                        pass
                    break
        except ReceiptHandleExpired as e:
            print(f"Receipt handle expired: {e.receipt_handle}")
            print("Job was likely redelivered to another worker")

    await pool.close()
```

--------------------------------

### Cancel In-Progress Job

Source: https://context7.com/adriangb/pgjobq/llms.txt

Demonstrates cancelling a job while it is actively being processed by a worker. This involves setting up a worker that acquires a job and then waits indefinitely, allowing the main task to cancel it.

```python
async def cancel_in_progress_job():
    """Cancel a job that's currently being processed"""
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("cancel-demo", pool)

    async with connect_to_queue("cancel-demo", pool) as queue:
        acquired = anyio.Event()

        async def worker():
            async with queue.receive() as job_stream:
                async for job_handle in job_stream:
                    async with job_handle.acquire() as job:
                        acquired.set()
                        print(f"Processing {job.id}...")
                        # Job will be cancelled while sleeping
                        await anyio.sleep(float("inf"))
                    return  # Exit after cancellation

        async with anyio.create_task_group() as tg:
            tg.start_soon(worker)
            async with queue.send(b'{"work": "important"}') as handle:
                await acquired.wait()  # Wait for worker to acquire job
                job_id = list(handle.jobs.keys())[0]
                await queue.cancel(job_id)
                print(f"Cancelled job {job_id}")
                await handle.wait()  # Job is now complete (cancelled)

    await pool.close()
```

--------------------------------

### Cancel Jobs by ID or Attribute

Source: https://context7.com/adriangb/pgjobq/llms.txt

Cancel pending or in-progress jobs using their ID or attribute filters. Requires importing asyncpg, anyio, and pgjobq components.

```python
import asyncpg
import anyio
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version, Attribute, OutgoingJob

async def cancel_jobs_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("cancellable-queue", pool)

    async with connect_to_queue("cancellable-queue", pool) as queue:
        # Send jobs
        async with queue.send(OutgoingJob(b'{"task": 1}', {"category": "batch1"})) as handle1:
            job_id = list(handle1.jobs.keys())[0]

            async with queue.send(OutgoingJob(b'{"task": 2}', {"category": "batch1"})):
                pass
            async with queue.send(OutgoingJob(b'{"task": 3}', {"category": "batch2"})):
                pass

            # Cancel by job ID
            await queue.cancel(job_id)
            print(f"Cancelled job {job_id}")

            # Cancel multiple jobs by ID
            # await queue.cancel(job_id1, job_id2, job_id3)

            # Cancel by filter - all jobs with category="batch1"
            await queue.cancel(Attribute("category").eq("batch1"))
            print("Cancelled all batch1 jobs")

    await pool.close()
```

--------------------------------

### Filter Jobs by IDs in Python

Source: https://context7.com/adriangb/pgjobq/llms.txt

Filter jobs by their UUIDs using the JobIdIn filter class. This is useful for receiving a specific set of jobs that have already been identified.

```python
import asyncpg
from uuid import UUID
from pgjobq import connect_to_queue, create_queue, migrate_to_latest_version, JobIdIn

async def job_id_filter_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("id-filter-queue", pool)

    async with connect_to_queue("id-filter-queue", pool) as queue:
        # Send some jobs and collect their IDs
        job_ids = []
        async with queue.send(b'{"job": 1}') as handle:
            job_ids.extend(handle.jobs.keys())
        async with queue.send(b'{"job": 2}') as handle:
            job_ids.extend(handle.jobs.keys())
        async with queue.send(b'{"job": 3}') as handle:
            # Don't include this one
            pass

        # Receive only specific jobs by ID
        filter = JobIdIn(ids=job_ids[:2])  # Only first two jobs

        async with queue.receive(filter=filter) as job_stream:
            count = 0
            async for job_handle in job_stream:
                async with job_handle.acquire() as job:
                    print(f"Received job {job.id}")
                    count += 1
                if count >= 2:
                    break

    await pool.close()
```

```python
JobIdIn(ids=job_ids[:2])
```

--------------------------------

### Delete a Queue

Source: https://context7.com/adriangb/pgjobq/llms.txt

Remove a queue and all its associated jobs from the database using the `delete_queue` function. The function returns `True` if the queue was deleted, and `False` if it did not exist.

```python
import asyncpg
from pgjobq import create_queue, delete_queue, migrate_to_latest_version

async def delete_queue_example():
    pool = await asyncpg.create_pool(
        "postgres://postgres:postgres@localhost/postgres"
    )
    await migrate_to_latest_version(pool)
    await create_queue("temporary-queue", pool)

    # Delete the queue
    deleted = await delete_queue("temporary-queue", pool)

    if deleted:
        print("Queue deleted successfully")
    else:
        print("Queue did not exist")

    await pool.close()
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.