异步sqlalchemy支持
asyncsa的Python项目详细描述
异步Q
asyncio sqlalchemy支持，带有async/await
添加异步sqlalchemy支持
安装
pip安装asyncsa
用法示例
importasynciofromsqlalchemyimportInteger,Column,String,ForeignKeyfromasyncsa.manager.async_pgimportPostgresManagerfromasyncsa.model.mixinimportMixinfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportrelationshipfromsqlalchemy.sqlimportselectdsn="postgres://postgres:test@localhost:2345/test"conf={'dsn':dsn}Base=declarative_base()Manager=PostgresManager(conf=conf)classUser(Mixin,Base):_manager=Manager__tablename__='test'id=Column(Integer,autoincrement=True,primary_key=True)name=Column(String(50))def__repr__(self):return"<User(name='%s', id='%s')>"%(self.name,self.id)classParent(Mixin,Base):__tablename__='parent'_manager=Managerid=Column(Integer,primary_key=True)child=relationship("Child",lazy='joined')classChild(Mixin,Base):_manager=Manager__tablename__='child'id=Column(Integer,primary_key=True)parent_id=Column(Integer,ForeignKey('parent.id'))parent=relationship("Parent",back_populates="child")if__name__=='__main__':loop=asyncio.get_event_loop()loop.run_until_complete(Manager.connect())result=loop.run_until_complete(Manager.get(User.objects([User.id]).join(Parent.__table__,Parent.id==User.id)))print(result)# result = loop.run_until_complete(Child.create_table())# print(result)# u = loop.run_until_complete(User.create(name='test'))# print('create', u)# print('get', loop.run_until_complete(User.get(name='test')))# print('all', loop.run_until_complete(User.all()))# u.name = 'aaa'# print('save', loop.run_until_complete(u.save()))
许可证
asyncnsq是在mit许可下提供的。
欢迎加入QQ群-->： 979659372
