本文整理汇总了Python中sqlalchemy.orm.Session类的典型用法代码示例。如果您正苦于以下问题:Python Session类的具体用法?Python Session怎么用?Python Session使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Session类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_one_to_many_on_m2o
def test_one_to_many_on_m2o(self):
Node, nodes = self.classes.Node, self.tables.nodes
mapper(Node, nodes, properties={
'children': relationship(Node,
backref=sa.orm.backref('parentnode',
remote_side=nodes.c.name,
passive_updates=False),
)})
sess = Session()
n1 = Node(name='n1')
sess.add(n1)
n2 = Node(name='n11', parentnode=n1)
n3 = Node(name='n12', parentnode=n1)
n4 = Node(name='n13', parentnode=n1)
sess.add_all([n2, n3, n4])
sess.commit()
n1.name = 'new n1'
sess.commit()
eq_(['new n1', 'new n1', 'new n1'],
[n.parent
for n in sess.query(Node).filter(
Node.name.in_(['n11', 'n12', 'n13']))])
开发者ID:ContextLogic,项目名称:sqlalchemy,代码行数:25,代码来源:test_naturalpks.py
示例2: test_orm_bundles
def test_orm_bundles(n):
"""Load lightweight "bundle" objects using the ORM."""
sess = Session(engine)
bundle = Bundle("customer", Customer.id, Customer.name, Customer.description)
for row in sess.query(bundle).yield_per(10000).limit(n):
pass
开发者ID:t3573393,项目名称:sqlalchemy,代码行数:7,代码来源:large_resultsets.py
示例3: test_illegal_operations
def test_illegal_operations(self):
User = self.classes.User
Address = self.classes.Address
s = Session()
for q, mname in (
(s.query(User).limit(2), r"limit\(\)"),
(s.query(User).offset(2), r"offset\(\)"),
(s.query(User).limit(2).offset(2), r"limit\(\)"),
(s.query(User).order_by(User.id), r"order_by\(\)"),
(s.query(User).group_by(User.id), r"group_by\(\)"),
(s.query(User).distinct(), r"distinct\(\)"),
(s.query(User).join(User.addresses),
r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)"),
(s.query(User).outerjoin(User.addresses),
r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)"),
(s.query(User).select_from(Address),
r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)"),
(s.query(User).from_self(),
r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)"),
):
assert_raises_message(
exc.InvalidRequestError,
r"Can't call Query.update\(\) or Query.delete\(\) when "
"%s has been called" % mname,
q.update,
{'name': 'ed'})
assert_raises_message(
exc.InvalidRequestError,
r"Can't call Query.update\(\) or Query.delete\(\) when "
"%s has been called" % mname,
q.delete)
开发者ID:CyberCollins,项目名称:sqlalchemy,代码行数:33,代码来源:test_update_delete.py
示例4: test_any_wpoly
def test_any_wpoly(self):
ParentThing, DataContainer, Job, SubJob = \
self.classes.ParentThing,\
self.classes.DataContainer,\
self.classes.Job,\
self.classes.SubJob
Job_P = with_polymorphic(Job, SubJob, aliased=True)
s = Session()
q = s.query(Job).join(DataContainer.jobs).\
filter(
DataContainer.jobs.of_type(Job_P).\
any(Job_P.id < Job.id)
)
self.assert_compile(q,
"SELECT job.id AS job_id, job.type AS job_type, "
"job.container_id "
"AS job_container_id "
"FROM data_container "
"JOIN job ON data_container.id = job.container_id "
"WHERE EXISTS (SELECT 1 "
"FROM (SELECT job.id AS job_id, job.type AS job_type, "
"job.container_id AS job_container_id, "
"subjob.id AS subjob_id, subjob.attr AS subjob_attr "
"FROM job LEFT OUTER JOIN subjob ON job.id = subjob.id) AS anon_1 "
"WHERE data_container.id = anon_1.job_container_id AND job.id > anon_1.job_id)"
)
开发者ID:afeide,项目名称:LuoYunCloud,代码行数:28,代码来源:test_of_type.py
示例5: test_invocation_per_mapper
def test_invocation_per_mapper(self):
"""test that BakedLazyLoader is getting invoked with the
"baked_select" lazy setting.
"""
User, Address = self._o2m_fixture(lazy="baked_select")
sess = Session()
q = sess.query(User).options(lazyload(User.addresses))
with mock.patch.object(BakedLazyLoader, "_emit_lazyload") as el:
u1 = q.first()
u1.addresses
# not invoked
eq_(el.mock_calls, [])
sess = Session()
q = sess.query(User)
with mock.patch.object(BakedLazyLoader, "_emit_lazyload") as el:
u1 = q.first()
u1.addresses
# invoked
is_(
el.mock_calls[0][1][1],
u1._sa_instance_state
)
开发者ID:CyberCollins,项目名称:sqlalchemy,代码行数:26,代码来源:test_baked.py
示例6: test_move_persistent_clean
def test_move_persistent_clean(self):
sess, u1 = self._persistent_fixture()
sess.close()
s2 = Session()
s2.add(u1)
self._assert_no_cycle(u1)
self._assert_not_modified(u1)
开发者ID:DeepakAlevoor,项目名称:sqlalchemy,代码行数:7,代码来源:test_session.py
示例7: test_set_composite_attrs_via_selectable
def test_set_composite_attrs_via_selectable(self):
Values, CustomValues, values, Descriptions, descriptions = (self.classes.Values,
self.classes.CustomValues,
self.tables.values,
self.classes.Descriptions,
self.tables.descriptions)
session = Session()
d = Descriptions(
custom_descriptions = CustomValues('Color', 'Number'),
values =[
Values(custom_values = CustomValues('Red', '5')),
Values(custom_values=CustomValues('Blue', '1'))
]
)
session.add(d)
session.commit()
eq_(
testing.db.execute(descriptions.select()).fetchall(),
[(1, 'Color', 'Number')]
)
eq_(
testing.db.execute(values.select()).fetchall(),
[(1, 1, 'Red', '5'), (2, 1, 'Blue', '1')]
)
开发者ID:Affirm,项目名称:sqlalchemy,代码行数:26,代码来源:test_composites.py
示例8: upgrade
def upgrade():
session = Session(bind=op.get_bind())
for i, item in enumerate(session.query(Salad).order_by(Salad.position)):
item.position = i
session.commit()
开发者ID:AlexFrazer,项目名称:sopython-site,代码行数:7,代码来源:20e76fbc8f6_fix_salad_position.py
示例9: test_not_supported_by_dialect_should_just_use_update
def test_not_supported_by_dialect_should_just_use_update(self):
User = self.classes.User
sess = Session()
self.assert_compile(sess.query(User.id).with_for_update(read=True),
"SELECT users.id AS users_id FROM users FOR UPDATE",
dialect=default.DefaultDialect()
)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:7,代码来源:test_lockmode.py
示例10: test_seven
def test_seven(self):
Parent, Base1, Base2, Sub1, Sub2, EP1, EP2 = self._classes()
s = Session()
self.assert_compile(
# adding Sub2 to the entities list helps it,
# otherwise the joins for Sub2.ep1/ep2 don't have columns
# to latch onto. Can't really make it better than this
s.query(Parent, Sub2).join(Parent.sub1).\
join(Sub1.sub2).from_self().\
join(Sub2.ep1).
join(Sub2.ep2),
"SELECT anon_1.parent_id AS anon_1_parent_id, "
"anon_1.parent_data AS anon_1_parent_data, "
"anon_1.sub2_id AS anon_1_sub2_id, "
"anon_1.base2_id AS anon_1_base2_id, "
"anon_1.base2_base1_id AS anon_1_base2_base1_id, "
"anon_1.base2_data AS anon_1_base2_data, "
"anon_1.sub2_subdata AS anon_1_sub2_subdata "
"FROM (SELECT parent.id AS parent_id, parent.data AS parent_data, "
"sub2.id AS sub2_id, "
"base2.id AS base2_id, "
"base2.base1_id AS base2_base1_id, "
"base2.data AS base2_data, "
"sub2.subdata AS sub2_subdata "
"FROM parent JOIN (base1 JOIN sub1 ON base1.id = sub1.id) "
"ON parent.id = sub1.parent_id JOIN "
"(base2 JOIN sub2 ON base2.id = sub2.id) "
"ON base1.id = base2.base1_id) AS anon_1 "
"JOIN ep1 ON anon_1.base2_id = ep1.base2_id "
"JOIN ep2 ON anon_1.base2_id = ep2.base2_id"
)
开发者ID:eevee,项目名称:sqlalchemy,代码行数:32,代码来源:test_relationship.py
示例11: _two_obj_fixture
def _two_obj_fixture(self):
e1 = Engineer(name='wally')
e2 = Engineer(name='dilbert', reports_to=e1)
sess = Session()
sess.add_all([e1, e2])
sess.commit()
return sess
开发者ID:eevee,项目名称:sqlalchemy,代码行数:7,代码来源:test_relationship.py
示例12: test_create_several_scenario_nodes
def test_create_several_scenario_nodes(fresh_database_config, scenario_manager):
# pylint: disable=invalid-name
"""Test that several scenario nodes can be created by library.
1. Create 2 scenario node with scenario manager.
2. Check type of the returned value.
3. Check that records for scenario node and node state are created.
"""
specs = [
NewScenarioNodeSpec(name=u'первый'),
NewScenarioNodeSpec(name=u'second'),
]
nodes = scenario_manager.create_scenario_nodes(specs)
assert len(nodes) == 2, "Wrong number of scenario nodes has been returned"
assert all(isinstance(node, ScenarioNode) for node in nodes), (
"Wrong type returned by create_scenario_node method"
)
engine = create_engine(fresh_database_config.get_connection_string())
session = Session(bind=engine)
node_records = session.query(ScenarioNodeRecord).all()
assert len(node_records) == 2, "Wrong number of scenario node records"
created_data = set()
for record in node_records:
state = session.query(ScenarioNodeStateRecord).filter_by(node_id=record.node_id).one()
created_data.add(state.name)
expected_data = set(spec.name for spec in specs)
assert created_data == expected_data, "Wrong states have been created in the database"
开发者ID:KillAChicken,项目名称:AutoStorage,代码行数:30,代码来源:test_scenario_node_creation.py
示例13: get
def get(session: orm.Session, discord_id=None, user_id=None, twitch_id=None, create=True):
"""
:param session:
:param discord_id:
:param user_id:
:param twitch_id:
:param create:
:rtype: roboto.model.User
:return:
"""
if not any([discord_id, user_id, twitch_id]):
return None
q = session.query(User)
if user_id:
col = User.user_id
val = user_id
elif twitch_id:
col = User.twitch_id
val = twitch_id.lower()
else:
col = User.discord_id
val = discord_id
q = q.filter(col == val).first()
if q:
return q
if create:
user = User()
setattr(user, col.key, val)
session.add(user)
log.debug("Creating new user: {}".format(user))
return user
return None
开发者ID:leighmacdonald,项目名称:twitch_markov,代码行数:33,代码来源:model.py
示例14: update_song_id3
def update_song_id3(engine, song_name, new_name='', new_artist='',
new_album=''):
session = Session(bind=engine)
song = session.query(Song).filter(Song.name == song_name)
try:
song_path = song.one().path
pass
except Exception:
song_path = './song.mp3'
updated = {}
if new_name:
updated['name'] = new_name
if new_artist:
updated['artist'] = new_artist
if new_album:
updated['album'] = new_album
song.update(updated)
session.commit()
original_song = EasyID3(song_path)
for key, value in updated.items():
if key == 'name':
key = 'title'
original_song[key] = value
original_song.save()
开发者ID:mileto94,项目名称:MusicPlayer,代码行数:27,代码来源:factory.py
示例15: test_batch_interaction
def test_batch_interaction(self):
"""test batching groups same-structured, primary
key present statements together.
"""
t = self.tables.t
class T(fixtures.ComparableEntity):
pass
mapper(T, t)
sess = Session()
sess.add_all([
T(data='t1'),
T(data='t2'),
T(id=3, data='t3'),
T(id=4, data='t4'),
T(id=5, data='t5'),
T(id=6, data=func.lower('t6')),
T(id=7, data='t7'),
T(id=8, data='t8'),
T(id=9, data='t9', def_='def2'),
T(id=10, data='t10', def_='def3'),
T(id=11, data='t11'),
])
self.assert_sql_execution(
testing.db,
sess.flush,
CompiledSQL(
"INSERT INTO t (data) VALUES (:data)",
{'data': 't1'}
),
CompiledSQL(
"INSERT INTO t (data) VALUES (:data)",
{'data': 't2'}
),
CompiledSQL(
"INSERT INTO t (id, data) VALUES (:id, :data)",
[{'data': 't3', 'id': 3},
{'data': 't4', 'id': 4},
{'data': 't5', 'id': 5}]
),
CompiledSQL(
"INSERT INTO t (id, data) VALUES (:id, lower(:lower_1))",
{'lower_1': 't6', 'id': 6}
),
CompiledSQL(
"INSERT INTO t (id, data) VALUES (:id, :data)",
[{'data': 't7', 'id': 7}, {'data': 't8', 'id': 8}]
),
CompiledSQL(
"INSERT INTO t (id, data, def_) VALUES (:id, :data, :def_)",
[{'data': 't9', 'id': 9, 'def_':'def2'},
{'data': 't10', 'id': 10, 'def_':'def3'}]
),
CompiledSQL(
"INSERT INTO t (id, data) VALUES (:id, :data)",
{'data': 't11', 'id': 11}
),
)
开发者ID:ContextLogic,项目名称:sqlalchemy,代码行数:60,代码来源:test_unitofworkv2.py
示例16: test_postgres_read
def test_postgres_read(self):
User = self.classes.User
sess = Session()
self.assert_compile(sess.query(User.id).with_for_update(read=True),
"SELECT users.id AS users_id FROM users FOR SHARE",
dialect="postgresql"
)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:7,代码来源:test_lockmode.py
示例17: test_is_modified_passive_on
def test_is_modified_passive_on(self):
User, Address = self._default_mapping_fixture()
s = Session()
u = User(name='fred', addresses=[Address(email_address='foo')])
s.add(u)
s.commit()
u.id
def go():
assert not s.is_modified(u, passive=True)
self.assert_sql_count(
testing.db,
go,
0
)
u.name = 'newname'
def go():
assert s.is_modified(u, passive=True)
self.assert_sql_count(
testing.db,
go,
0
)
开发者ID:DeepakAlevoor,项目名称:sqlalchemy,代码行数:25,代码来源:test_session.py
示例18: test_postgres_update_of_entity
def test_postgres_update_of_entity(self):
User = self.classes.User
sess = Session()
self.assert_compile(sess.query(User.id).with_for_update(of=User),
"SELECT users.id AS users_id FROM users FOR UPDATE OF users",
dialect="postgresql"
)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:7,代码来源:test_lockmode.py
示例19: test_not_none
def test_not_none(self):
Graph, Edge, Point = (self.classes.Graph,
self.classes.Edge,
self.classes.Point)
# current contract. the composite is None
# when hasn't been populated etc. on a
# pending/transient object.
e1 = Edge()
assert e1.end is None
sess = Session()
sess.add(e1)
# however, once it's persistent, the code as of 0.7.3
# would unconditionally populate it, even though it's
# all None. I think this usage contract is inconsistent,
# and it would be better that the composite is just
# created unconditionally in all cases.
# but as we are just trying to fix [ticket:2308] and
# [ticket:2309] without changing behavior we maintain
# that only "persistent" gets the composite with the
# Nones
sess.flush()
assert e1.end is not None
开发者ID:Affirm,项目名称:sqlalchemy,代码行数:25,代码来源:test_composites.py
示例20: test_postgres_for_no_key_nowait_update
def test_postgres_for_no_key_nowait_update(self):
User = self.classes.User
sess = Session()
self.assert_compile(sess.query(User.id).with_for_update(key_share=True, nowait=True),
"SELECT users.id AS users_id FROM users FOR NO KEY UPDATE NOWAIT",
dialect="postgresql"
)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:7,代码来源:test_lockmode.py
注:本文中的sqlalchemy.orm.Session类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论