import asyncio
from taskiq_postgresql import PostgresqlBroker, PostgresqlResultBackend

# Configure the result backend
result_backend = PostgresqlResultBackend(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db",
)

# Configure the broker with result backend
broker = PostgresqlBroker(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db",
).with_result_backend(result_backend)


@broker.task
async def calculate_sum(a: int, b: int) -> int:
    """Calculate the sum of two numbers."""
    await asyncio.sleep(1)  # Simulate some work
    return a + b


async def main():
    # Startup the broker
    await broker.startup()
    
    # Send a task
    task = await calculate_sum.kiq(10, 20)
    
    # Wait for result
    result = await task.wait_result()
    print(f"Result: {result}")  # Result: 30
    
    # Shutdown the broker
    await broker.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

from taskiq_postgresql import PostgresqlBroker, PostgresqlSchedulerSource
from taskiq import TaskiqScheduler

# Initialize broker
broker = PostgresqlBroker(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db"
)

# Initialize scheduler source
scheduler_source = PostgresqlSchedulerSource(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db",
    table_name="taskiq_schedules",
    driver="asyncpg"
)

# Create scheduler
scheduler = TaskiqScheduler(
    broker=broker,
    sources=[scheduler_source],
)

@broker.task
async def scheduled_task():
    print("This task runs on schedule!")

# Schedule task to run every minute
async def setup_schedule():
    await scheduler_source.add_schedule(
        schedule_id="task-every-minute",
        task_name="scheduled_task",
        cron="* * * * *",  # Every minute
        args=[],
        kwargs={}
    )

from taskiq.serializers import JSONSerializer

result_backend = PostgresqlResultBackend(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db",
    serializer=JSONSerializer(),
)

broker = PostgresqlBroker(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db",
    queue_name="my_custom_queue",
    field_for_task_id="Text",  # Use TEXT instead of UUID
)

result_backend = PostgresqlResultBackend(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db",
    table_name="my_custom_results",
    field_for_task_id="VarChar",  # Use VARCHAR instead of UUID
)

# AsyncPG
broker = PostgresqlBroker(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db",
    driver="asyncpg",
    min_size=5,
    max_size=20,
    max_inactive_connection_lifetime=300,
)

# Psycopg3
broker = PostgresqlBroker(
    dsn="postgresql://postgres:postgres@localhost:5432/taskiq_db",
    driver="psycopg",
    min_size=5,
    max_size=20,
    max_lifetime=3600,
)

import os

# From environment
dsn = os.getenv("DATABASE_URL", "postgresql://localhost/taskiq")

# With SSL
dsn = "postgresql://user:pass@localhost:5432/db?sslmode=require"

broker = PostgresqlBroker(dsn=dsn)

# Ensure your connection string is correct
dsn = "postgresql://username:password@host:port/database"

# Check PostgreSQL is running and accessible
import asyncpg
conn = await asyncpg.connect(dsn)
await conn.close()

# Ensure user has CREATE TABLE permissions
# Or manually create tables using provided schemas