我们从Python开源项目中，提取了以下8个代码示例，用于说明如何使用sqlalchemy.orm.create_session()。
def test_only_loads_relations_on_unpopulated_models(self): User = self.classes.User Address = self.classes.Address session = create_session() users = session.query(User).order_by(self.tables.users.c.id.asc()).all() address = session.query(Address).filter(self.tables.addresses.c.id == 1).first() # pre-load the address for the first user attributes.set_committed_value(users[0], 'addresses', [address]) self.queries = [] # make sure no relations are loaded for user in users[1:]: model_dict = attributes.instance_dict(user) assert 'addresses' not in model_dict # trigger a lazy load users[1].addresses # only 1 query should have been generated to load all the child relationships assert len(self.queries) == 1 unpopulated_user_ids = [user.id for user in users[1:]] assert self.queries[0]['parameters'] == tuple(unpopulated_user_ids)
def test_load_one_to_one(self): User = self.classes.User UserInfo = self.classes.UserInfo session = create_session() users = session.query(User).order_by(self.tables.users.c.id.asc()).all() self.queries = [] # make sure no relations are loaded for user in users: model_dict = attributes.instance_dict(user) assert 'user_info' not in model_dict # trigger a lazy load on the first user users[0].user_info # only 1 query should have been generated to load all the child relationships assert len(self.queries) == 1 self.queries = [] user1_dict = attributes.instance_dict(users[0]) user2_dict = attributes.instance_dict(users[1]) user3_dict = attributes.instance_dict(users[2]) user4_dict = attributes.instance_dict(users[3]) assert UserInfo(id=1, details='is cool', user_id=7) == user1_dict['user_info'] assert UserInfo(id=2, details='is not cool', user_id=8) == user2_dict['user_info'] assert None == user3_dict['user_info'] assert UserInfo(id=3, details='is moderately cool', user_id=10) == user4_dict['user_info'] # backrefs should also not trigger loading assert users[0].user_info.user == users[0] assert users[1].user_info.user == users[1] assert users[3].user_info.user == users[3] # no new queries should have been generated assert len(self.queries) == 0
def test_load_many_to_one(self): User = self.classes.User Address = self.classes.Address session = create_session() addresses = session.query(Address).order_by(self.tables.addresses.c.id.asc()).all() self.queries = [] # make sure no relations are loaded for address in addresses: model_dict = attributes.instance_dict(address) assert 'user' not in model_dict # trigger a lazy load on the first user addresses[0].user # only 1 query should have been generated to load all the child relationships assert len(self.queries) == 1 self.queries = [] address1_dict = attributes.instance_dict(addresses[0]) address2_dict = attributes.instance_dict(addresses[1]) address3_dict = attributes.instance_dict(addresses[2]) address4_dict = attributes.instance_dict(addresses[3]) address5_dict = attributes.instance_dict(addresses[4]) assert User(id=7, name='jack', parent_id=None) == address1_dict['user'] assert User(id=8, name='jack jr', parent_id=7) == address2_dict['user'] assert User(id=8, name='jack jr', parent_id=7) == address3_dict['user'] assert User(id=8, name='jack jr', parent_id=7) == address4_dict['user'] assert User(id=9, name='fred', parent_id=7) == address5_dict['user'] # no new queries should have been generated assert len(self.queries) == 0
def test_load_many_to_many(self): User = self.classes.User Thing = self.classes.Thing session = create_session() users = session.query(User).order_by(self.tables.users.c.id.asc()).all() self.queries = [] # make sure no relations are loaded for user in users: model_dict = attributes.instance_dict(user) assert 'things' not in model_dict # trigger a lazy load on the first user users[0].things # only 1 query should have been generated to load all the child relationships assert len(self.queries) == 1 self.queries = [] user1_dict = attributes.instance_dict(users[0]) user2_dict = attributes.instance_dict(users[1]) user3_dict = attributes.instance_dict(users[2]) user4_dict = attributes.instance_dict(users[3]) assert [ Thing(id=1, name='dog'), ] == user1_dict['things'] assert [ Thing(id=1, name='dog'), ] == user2_dict['things'] assert [ Thing(id=2, name='lamp'), ] == user3_dict['things'] assert [ Thing(id=2, name='lamp'), Thing(id=3, name='chair'), ] == user4_dict['things'] # no new queries should have been generated assert len(self.queries) == 0
def _connect(self, url): self.url = url # TODO: seems like 0.5.x branch does not work with engine.dispose and staticpool #self.engine = create_engine(url, echo=True, poolclass=StaticPool) self.engine = create_engine(url, echo=True) # silence the logger added by SA, nose adds its own! logging.getLogger('sqlalchemy').handlers=[] self.meta = MetaData(bind=self.engine) if self.level < self.CONNECT: return #self.session = create_session(bind=self.engine) if self.level < self.TXN: return #self.txn = self.session.begin()
def SP(self): try: from sqlalchemy.orm import create_session session = create_session() connection = connect.ConnectorMySql().raw_connection() cursor = connection.cursor() cursor.callproc("GetReportersOfSite", [None,None]) results = list(cursor.fetchall()) cursor.close() connection.commit() except Exception as e: raise Exception('Type Error In SP :'+ str(e))
def test_load_one_to_many_self_refencing(self): User = self.classes.User session = create_session() users = ( session.query(User) .filter(self.tables.users.c.id.in_([7,8])) .order_by(self.tables.users.c.id.asc()) .all() ) self.queries = [] # make sure no relations are loaded for user in users: model_dict = attributes.instance_dict(user) assert 'children' not in model_dict # trigger a lazy load on the first user users[0].children # only 1 query should have been generated to load all the child relationships assert len(self.queries) == 1 self.queries = [] user1_dict = attributes.instance_dict(users[0]) user2_dict = attributes.instance_dict(users[1]) assert [ User(id=8, name='jack jr', parent_id=7), User(id=9, name='fred', parent_id=7), ] == user1_dict['children'] assert [ User(id=10, name='jack jr jr', parent_id=8), ] == user2_dict['children'] # backrefs should also not trigger loading for child in users[0].children: assert child.parent == users[0] for child in users[1].children: assert child.parent == users[1] # no new queries should have been generated assert len(self.queries) == 0
