sqlalchemy_async_orm_model_crud_with_aiosqlite.py

1import asyncio
2
3from sqlalchemy import Column
4from sqlalchemy import DateTime
5from sqlalchemy import ForeignKey
6from sqlalchemy import Integer
7from sqlalchemy import String
8from sqlalchemy import func
9from sqlalchemy import select
10from sqlalchemy.ext.asyncio import AsyncSession
11from sqlalchemy.ext.asyncio import create_async_engine
12from sqlalchemy.orm import declarative_base
13from sqlalchemy.orm import relationship
14from sqlalchemy.orm import selectinload
15from sqlalchemy.orm import sessionmaker
16
17Base = declarative_base()
18
19
20class A(Base):
21    __tablename__ = "a"
22
23    id = Column(Integer, primary_key=True)
24    data = Column(String)
25    create_date = Column(DateTime, server_default=func.now())
26    bs = relationship("B")
27
28    def __repr__(self):
29        return f"A(id={self.id!r}, data={self.data!r}, create_date={self.create_date!r})"
30
31
32class B(Base):
33    __tablename__ = "b"
34
35    id = Column(Integer, primary_key=True)
36    a_id = Column(ForeignKey("a.id"))
37    data = Column(String)
38
39    def __repr__(self):
40        return f"B(id={self.id!r}, data={self.data!r})"
41
42
43async def async_main():
44    engine = create_async_engine(
45        "sqlite+aiosqlite://",
46        echo=True,
47    )
48
49    async with engine.begin() as conn:
50        await conn.run_sync(Base.metadata.drop_all)
51        await conn.run_sync(Base.metadata.create_all)
52
53    # expire_on_commit=False will prevent attributes from being expired
54    # after commit.
55    async_session = sessionmaker(
56        engine, expire_on_commit=False, class_=AsyncSession
57    )
58
59    async with async_session() as session:
60        async with session.begin():
61            session.add_all(
62                [
63                    A(data="a1", bs=[B(data="b1"), B(data="b2")]),
64                    A(data="a2", bs=[B(data="b3"), B(data="b4")]),
65                ]
66            )
67
68        # for SQLAlchemy 2.0 style, use session.execute()
69        stmt = select(A).options(selectinload(A.bs))
70
71        result = await session.execute(stmt)
72
73        for a1 in result.scalars():
74            print(a1)
75            print(f"created at: {a1.create_date}")
76            for b1 in a1.bs:
77                print(b1)
78
79        result = await session.execute(select(A).order_by(A.id))
80
81        a1 = result.scalars().first()
82
83        a1.data = "new data"
84
85        await session.commit()
86
87    # for AsyncEngine created in function scope, close and
88    # clean up pooled connections
89    await engine.dispose()
90
91
92asyncio.run(async_main())