41 lines
1.0 KiB
Python
41 lines
1.0 KiB
Python
"""
|
|
Async example using the provided async engine/session and generic AsyncRepository.
|
|
|
|
Run: uv run python examples/04_async_crud.py
|
|
"""
|
|
|
|
import asyncio
|
|
from typing import Optional
|
|
from sqlmodel import select, SQLModel, Field
|
|
|
|
from sqlmodel_pg_kit import create_all, AsyncRepository
|
|
from sqlmodel_pg_kit.db import get_async_session
|
|
|
|
|
|
class Hero(SQLModel, table=True):
|
|
id: Optional[int] = Field(default=None, primary_key=True)
|
|
name: str = Field(index=True)
|
|
age: Optional[int] = None
|
|
|
|
|
|
async def amain():
|
|
# Ensure tables exist (sync helper is fine to call once)
|
|
create_all()
|
|
|
|
repo = AsyncRepository(Hero)
|
|
|
|
# Clean and insert using async session
|
|
async with get_async_session() as s:
|
|
await s.execute(Hero.__table__.delete())
|
|
await s.commit()
|
|
|
|
await repo.create(s, {"name": "Async Hero", "age": 7})
|
|
|
|
res = await s.execute(select(Hero))
|
|
heroes = res.scalars().all()
|
|
print([h.name for h in heroes])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(amain())
|