asyncsa 2.0.2

Description:

asyncsa 2.0.2

asyncsa


asyncio sqlalchemy support,with async/await
add asyncio sqlalchemy support
Install

pip install asyncsa
Usage examples

import asyncio
from sqlalchemy import Integer, Column, String, ForeignKey
from asyncsa.manager.async_pg import PostgresManager
from asyncsa.model.mixin import Mixin
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.sql import select

dsn = "postgres://postgres:test@localhost:2345/test"

conf = {'dsn': dsn}

Base = declarative_base()


Manager = PostgresManager(conf=conf)


class User(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)


class Parent(Mixin, Base):
__tablename__ = 'parent'
_manager = Manager
id = Column(Integer, primary_key=True)
child = relationship("Child", lazy='joined')


class Child(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()))

License
The asyncnsq is offered under APGLV3 license.