"""Illustrates use of the sqlalchemy.ext.asyncio.AsyncSession objectfor asynchronous ORM use, including the optional run_sync() method."""importasynciofromsqlalchemyimportColumnfromsqlalchemyimportForeignKeyfromsqlalchemyimportIntegerfromsqlalchemyimportStringfromsqlalchemy.ext.asyncioimportAsyncAttrsfromsqlalchemy.ext.asyncioimportAsyncSessionfromsqlalchemy.ext.asyncioimportcreate_async_enginefromsqlalchemy.futureimportselectfromsqlalchemy.ormimportDeclarativeBasefromsqlalchemy.ormimportrelationshipclassBase(AsyncAttrs,DeclarativeBase):passclassA(Base):__tablename__="a"id=Column(Integer,primary_key=True)data=Column(String)bs=relationship("B")classB(Base):__tablename__="b"id=Column(Integer,primary_key=True)a_id=Column(ForeignKey("a.id"))data=Column(String)defrun_queries(session):"""A function written in "synchronous" style that will be invoked within the asyncio event loop. The session object passed is a traditional orm.Session object with synchronous interface. """stmt=select(A)result=session.execute(stmt)fora1inresult.scalars():print(a1)# lazy loadsforb1ina1.bs:print(b1)result=session.execute(select(A).order_by(A.id))a1=result.scalars().first()a1.data="new data"asyncdefasync_main():"""Main program function."""engine=create_async_engine("postgresql+asyncpg://scott:tiger@localhost/test",echo=True,)asyncwithengine.begin()asconn:awaitconn.run_sync(Base.metadata.drop_all)awaitconn.run_sync(Base.metadata.create_all)asyncwithAsyncSession(engine)assession:asyncwithsession.begin():session.add_all([A(bs=[B(),B()],data="a1"),A(bs=[B()],data="a2"),A(bs=[B(),B()],data="a3"),])# we have the option to run a function written in sync style# within the AsyncSession.run_sync() method. The function will# be passed a synchronous-style Session object and the function# can use traditional ORM patterns.awaitsession.run_sync(run_queries)awaitsession.commit()asyncio.run(async_main())