simple asyncpg

gistfile1.txt

import asyncpg
import asyncio
import os
from dotenv import load_dotenv

load_dotenv()

async def run_test():
    db_url = os.getenv("DB_URL")
    if not db_url:
        raise ValueError("Database URL not provided")

    conn = None
    try:
        print("Connecting to database...", db_url)
        conn = await asyncpg.connect(db_url,
                                   command_timeout=60)
        print("Database connection established.")

        query = "SELECT * FROM tenants"
        result = await conn.fetch(query)
        print(result)
        print(f"Result type: {type(result)}")
        print(f"Result length: {len(result)}")

        if not result:
            print("No records found.")
        else:
            for row in result:
                print(row)

        # Create a table
        await conn.execute('''
            CREATE TABLE IF NOT EXISTS test_asyncpg(
                id SERIAL PRIMARY KEY,
                name TEXT,
                age INT
            )
        ''')
    
        # Insert a row
        await conn.execute('''
            INSERT INTO test_asyncpg(name, age) VALUES($1, $2)
        ''', 'Alice', 30)
    
        # Fetch rows
        rows = await conn.fetch('SELECT * FROM test_asyncpg')
        for row in rows:
            print(dict(row))  # convert asyncpg.Record to dict for easier reading

    except Exception as e:
        print(f"Error executing query: {e}")

    finally:
        if conn:
            await conn.close()
            print("Database connection closed.")

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