Akses Pangkalan Data Tak Segerak

import asyncio
import asyncpg

async def main():
    conn = None
    try:
        conn = await asyncpg.connect(user='postgres', password='mysecretpassword',
                                     database='testdb', host='localhost')
        print("Connected to PostgreSQL!")
        
        # Execute a query
        result = await conn.fetchval('SELECT 1 + 1')
        print(f"Query result: {result}") # Should be 2
        
    except Exception as e:
        print(f"Error: {e}")
    finally:
        if conn:
            await conn.close()
            print("Connection closed.")

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

from sqlmodel import create_engine, SQLModel
import asyncio

# Replace with your actual async PostgreSQL connection string
# The 'postgresql+asyncpg' part tells SQLAlchemy to use asyncpg
DATABASE_URL = "postgresql+asyncpg://postgres:mysecretpassword@localhost/testdb"

async def main():
    engine = create_engine(DATABASE_URL, echo=True)
    print("Async SQLModel engine created.")
    
    # In a real app, you'd usually create tables here
    # async with engine.begin() as conn:
    #     await conn.run_sync(SQLModel.metadata.create_all)
    
    # Just demonstrating engine creation for this example
    await engine.dispose()
    print("Engine disposed.")

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

from typing import Optional
from sqlmodel import Field, SQLModel

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

# This model can now be used with FastAPI for request bodies
# and with SQLAlchemy for database interactions.
print("Hero model defined successfully!")

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Session, select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
import asyncio

DATABASE_URL = "postgresql+asyncpg://postgres:mysecretpassword@localhost/testdb"
async_engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=async_engine, class_=AsyncSession)

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

async def create_db_and_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    print("Database tables created/updated.")

async def create_hero(hero: Hero):
    async with AsyncSessionLocal() as session:
        session.add(hero)
        await session.commit()
        await session.refresh(hero)
        print(f"Created hero: {hero.name} (ID: {hero.id})")
        return hero

async def get_heroes():
    async with AsyncSessionLocal() as session:
        statement = select(Hero)
        results = await session.execute(statement)
        heroes = results.scalars().all()
        print("\nAll Heroes:")
        for hero in heroes:
            print(f"- {hero.name} (Age: {hero.age})")
        return heroes

async def main():
    await create_db_and_tables()

    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", age=28)
    hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
    hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)

    await create_hero(hero_1)
    await create_hero(hero_2)
    await create_hero(hero_3)

    await get_heroes()
    await async_engine.dispose()

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