import sqlalchemy as sa

from sqlalchemy import event

from sqlalchemy import ForeignKey

from sqlalchemy import inspect

from sqlalchemy import Integer

from sqlalchemy import select

from sqlalchemy import Sequence

from sqlalchemy import String

from sqlalchemy import testing

from sqlalchemy.orm import attributes

from sqlalchemy.orm import backref

from sqlalchemy.orm import close_all_sessions

from sqlalchemy.orm import create_session

from sqlalchemy.orm import exc as orm_exc

from sqlalchemy.orm import joinedload

from sqlalchemy.orm import make_transient

from sqlalchemy.orm import make_transient_to_detached

from sqlalchemy.orm import mapper

from sqlalchemy.orm import object_session

from sqlalchemy.orm import relationship

from sqlalchemy.orm import Session

from sqlalchemy.orm import sessionmaker

from sqlalchemy.orm import was_deleted

from sqlalchemy.testing import assert_raises

from sqlalchemy.testing import assert_raises_message

from sqlalchemy.testing import assertions

from sqlalchemy.testing import config

from sqlalchemy.testing import engines

from sqlalchemy.testing import eq_

from sqlalchemy.testing import fixtures

from sqlalchemy.testing import is_

from sqlalchemy.testing import is_not_

from sqlalchemy.testing import is_true

from sqlalchemy.testing import mock

from sqlalchemy.testing import pickleable

from sqlalchemy.testing.schema import Column

from sqlalchemy.testing.schema import Table

from sqlalchemy.testing.util import gc_collect

from sqlalchemy.util import pickle

from sqlalchemy.util.compat import inspect_getfullargspec

from test.orm import _fixtures

class ExecutionTest(_fixtures.FixtureTest):

run_inserts = None

__backend__ = True

@testing.requires.sequences

def test_sequence_execute(self):

seq = Sequence("some_sequence")

seq.create(testing.db)

try:

sess = create_session(bind=testing.db)

eq_(sess.execute(seq), 1)

finally:

seq.drop(testing.db)

def test_textual_execute(self):

"""test that Session.execute() converts to text()"""

users = self.tables.users

sess = create_session(bind=self.metadata.bind)

users.insert().execute(id=7, name="jack")

# use :bindparam style

eq_(

sess.execute(

"select * from users where id=:id", {"id": 7}

).fetchall(),

[(7, "jack")],

)

# use :bindparam style

eq_(sess.scalar("select id from users where id=:id", {"id": 7}), 7)

def test_parameter_execute(self):

users = self.tables.users

sess = Session(bind=testing.db)

sess.execute(

users.insert(), [{"id": 7, "name": "u7"}, {"id": 8, "name": "u8"}]

)

sess.execute(users.insert(), {"id": 9, "name": "u9"})

eq_(

sess.execute(

sa.select([users.c.id]).order_by(users.c.id)

).fetchall(),

[(7,), (8,), (9,)],

)

class TransScopingTest(_fixtures.FixtureTest):

run_inserts = None

__prefer_requires__ = ("independent_connections",)

def test_no_close_on_flush(self):

"""Flush() doesn't close a connection the session didn't open"""

User, users = self.classes.User, self.tables.users

c = testing.db.connect()

c.exec_driver_sql("select * from users")

mapper(User, users)

s = create_session(bind=c)

s.add(User(name="first"))

s.flush()

c.exec_driver_sql("select * from users")

def test_close(self):

"""close() doesn't close a connection the session didn't open"""

User, users = self.classes.User, self.tables.users

c = testing.db.connect()

c.exec_driver_sql("select * from users")

mapper(User, users)

s = create_session(bind=c)

s.add(User(name="first"))

s.flush()

c.exec_driver_sql("select * from users")

s.close()

c.exec_driver_sql("select * from users")

def test_autobegin_execute(self):

# test the new autobegin behavior introduced in #5074

s = Session(testing.db)

is_(s._transaction, None)

s.execute(select([1]))

is_not_(s._transaction, None)

s.commit()

is_(s._transaction, None)

s.execute(select([1]))

is_not_(s._transaction, None)

s.close()

is_(s._transaction, None)

s.execute(select([1]))

is_not_(s._transaction, None)

s.close()

is_(s._transaction, None)

def test_autobegin_flush(self):

# test the new autobegin behavior introduced in #5074

User, users = self.classes.User, self.tables.users

mapper(User, users)

s = Session(testing.db)

is_(s._transaction, None)

# empty flush, nothing happens

s.flush()

is_(s._transaction, None)

s.add(User(id=1, name="name"))

s.flush()

is_not_(s._transaction, None)

s.commit()

is_(s._transaction, None)

def test_autobegin_begin_method(self):

s = Session(testing.db)

s.begin() # OK

assert_raises_message(

sa.exc.InvalidRequestError,

"A transaction is already begun. Use "

"subtransactions=True to allow subtransactions.",

s.begin,

)

@testing.requires.independent_connections

@engines.close_open_connections

def test_transaction(self):

User, users = self.classes.User, self.tables.users

mapper(User, users)

conn1 = testing.db.connect()

conn2 = testing.db.connect()

sess = create_session(autocommit=False, bind=conn1)

u = User(name="x")

sess.add(u)

sess.flush()

assert (

conn1.exec_driver_sql("select count(1) from users").scalar() == 1

)

assert (

conn2.exec_driver_sql("select count(1) from users").scalar() == 0

)

sess.commit()

assert (

conn1.exec_driver_sql("select count(1) from users").scalar() == 1

)

assert (

testing.db.connect()

.exec_driver_sql("select count(1) from users")

.scalar()

== 1

)

sess.close()

class SessionUtilTest(_fixtures.FixtureTest):

run_inserts = None

def test_close_all_sessions(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

s1 = Session()

u1 = User()

s1.add(u1)

s2 = Session()

u2 = User()

s2.add(u2)

assert u1 in s1

assert u2 in s2

close_all_sessions()

assert u1 not in s1

assert u2 not in s2

def test_session_close_all_deprecated(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

s1 = Session()

u1 = User()

s1.add(u1)

s2 = Session()

u2 = User()

s2.add(u2)

assert u1 in s1

assert u2 in s2

with assertions.expect_deprecated(

r"The Session.close_all\(\) method is deprecated and will "

"be removed in a future release. "

):

Session.close_all()

assert u1 not in s1

assert u2 not in s2

def test_object_session_raises(self):

User = self.classes.User

assert_raises(orm_exc.UnmappedInstanceError, object_session, object())

assert_raises(orm_exc.UnmappedInstanceError, object_session, User())

def test_make_transient(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = create_session()

sess.add(User(name="test"))

sess.flush()

u1 = sess.query(User).first()

make_transient(u1)

assert u1 not in sess

sess.add(u1)

assert u1 in sess.new

u1 = sess.query(User).first()

sess.expunge(u1)

make_transient(u1)

sess.add(u1)

assert u1 in sess.new

# test expired attributes

# get unexpired

u1 = sess.query(User).first()

sess.expire(u1)

make_transient(u1)

assert u1.id is None

assert u1.name is None

# works twice

make_transient(u1)

sess.close()

u1.name = "test2"

sess.add(u1)

sess.flush()

assert u1 in sess

sess.delete(u1)

sess.flush()

assert u1 not in sess

assert_raises(sa.exc.InvalidRequestError, sess.add, u1)

make_transient(u1)

sess.add(u1)

sess.flush()

assert u1 in sess

def test_make_transient_plus_rollback(self):

# test for [ticket:2182]

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

u1 = User(name="test")

sess.add(u1)

sess.commit()

sess.delete(u1)

sess.flush()

make_transient(u1)

sess.rollback()

assert attributes.instance_state(u1).transient

def test_make_transient_to_detached(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

u1 = User(id=1, name="test")

sess.add(u1)

sess.commit()

sess.close()

u2 = User(id=1)

make_transient_to_detached(u2)

assert "id" in u2.__dict__

sess.add(u2)

eq_(u2.name, "test")

def test_make_transient_to_detached_no_session_allowed(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

u1 = User(id=1, name="test")

sess.add(u1)

assert_raises_message(

sa.exc.InvalidRequestError,

"Given object must be transient",

make_transient_to_detached,

u1,

)

def test_make_transient_to_detached_no_key_allowed(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

u1 = User(id=1, name="test")

sess.add(u1)

sess.commit()

sess.expunge(u1)

assert_raises_message(

sa.exc.InvalidRequestError,

"Given object must be transient",

make_transient_to_detached,

u1,

)

class SessionStateTest(_fixtures.FixtureTest):

run_inserts = None

__prefer_requires__ = ("independent_connections",)

def test_info(self):

s = Session()

eq_(s.info, {})

maker = sessionmaker(info={"global": True, "s1": 5})

s1 = maker()

s2 = maker(info={"s1": 6, "s2": True})

eq_(s1.info, {"global": True, "s1": 5})

eq_(s2.info, {"global": True, "s1": 6, "s2": True})

s2.info["global"] = False

s2.info["s1"] = 7

s3 = maker()

eq_(s3.info, {"global": True, "s1": 5})

maker2 = sessionmaker()

s4 = maker2(info={"s4": 8})

eq_(s4.info, {"s4": 8})

@testing.requires.independent_connections

@engines.close_open_connections

def test_autoflush(self):

User, users = self.classes.User, self.tables.users

bind = self.metadata.bind

mapper(User, users)

conn1 = bind.connect()

conn2 = bind.connect()

sess = create_session(bind=conn1, autocommit=False, autoflush=True)

u = User()

u.name = "ed"

sess.add(u)

u2 = sess.query(User).filter_by(name="ed").one()

assert u2 is u

eq_(conn1.exec_driver_sql("select count(1) from users").scalar(), 1)

eq_(conn2.exec_driver_sql("select count(1) from users").scalar(), 0)

sess.commit()

eq_(conn1.exec_driver_sql("select count(1) from users").scalar(), 1)

eq_(

bind.connect()

.exec_driver_sql("select count(1) from users")

.scalar(),

1,

)

sess.close()

def test_with_no_autoflush(self):

User, users = self.classes.User, self.tables.users

mapper(User, users)

sess = Session()

u = User()

u.name = "ed"

sess.add(u)

def go(obj):

assert u not in sess.query(User).all()

testing.run_as_contextmanager(sess.no_autoflush, go)

assert u in sess.new

assert u in sess.query(User).all()

assert u not in sess.new

def test_with_no_autoflush_after_exception(self):

sess = Session(autoflush=True)

assert_raises(

ZeroDivisionError,

testing.run_as_contextmanager,

sess.no_autoflush,

lambda obj: 1 / 0,

)

is_true(sess.autoflush)

def test_autoflush_exception_addition(self):

User, users = self.classes.User, self.tables.users

Address, addresses = self.classes.Address, self.tables.addresses

mapper(User, users, properties={"addresses": relationship(Address)})

mapper(Address, addresses)

s = Session(testing.db)

u1 = User(name="first")

s.add(u1)

s.commit()

u1.addresses.append(Address(email=None))

# will raise for null email address

assert_raises_message(

sa.exc.DBAPIError,

".*raised as a result of Query-invoked autoflush; consider using "

"a session.no_autoflush block.*",

s.query(User).first,

)

def test_deleted_flag(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = sessionmaker()()

u1 = User(name="u1")

sess.add(u1)

sess.commit()

sess.delete(u1)

sess.flush()

assert u1 not in sess

assert_raises(sa.exc.InvalidRequestError, sess.add, u1)

assert sess.transaction is not None

sess.rollback()

assert u1 in sess

sess.delete(u1)

sess.commit()

assert u1 not in sess

assert_raises(sa.exc.InvalidRequestError, sess.add, u1)

make_transient(u1)

sess.add(u1)

sess.commit()

eq_(sess.query(User).count(), 1)

@testing.requires.sane_rowcount

def test_deleted_adds_to_imap_unconditionally(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

u1 = User(name="u1")

sess.add(u1)

sess.commit()

sess.delete(u1)

sess.flush()

# object is not in session

assert u1 not in sess

# but it *is* attached

assert u1._sa_instance_state.session_id == sess.hash_key

# mark as deleted again

sess.delete(u1)

# in the session again

assert u1 in sess

# commit proceeds w/ warning

with assertions.expect_warnings(

"DELETE statement on table 'users' "

r"expected to delete 1 row\(s\); 0 were matched."

):

sess.commit()

@testing.requires.independent_connections

@engines.close_open_connections

def test_autoflush_unbound(self):

User, users = self.classes.User, self.tables.users

mapper(User, users)

try:

sess = create_session(autocommit=False, autoflush=True)

u = User()

u.name = "ed"

sess.add(u)

u2 = sess.query(User).filter_by(name="ed").one()

assert u2 is u

assert (

sess.execute(

"select count(1) from users", mapper=User

).scalar()

== 1

)

assert (

testing.db.connect()

.exec_driver_sql("select count(1) from users")

.scalar()

== 0

)

sess.commit()

assert (

sess.execute(

"select count(1) from users", mapper=User

).scalar()

== 1

)

assert (

testing.db.connect()

.exec_driver_sql("select count(1) from users")

.scalar()

== 1

)

sess.close()

except Exception:

sess.rollback()

raise

@engines.close_open_connections

def test_autoflush_2(self):

User, users = self.classes.User, self.tables.users

mapper(User, users)

conn1 = testing.db.connect()

sess = create_session(bind=conn1, autocommit=False, autoflush=True)

u = User()

u.name = "ed"

sess.add(u)

sess.commit()

assert (

conn1.exec_driver_sql("select count(1) from users").scalar() == 1

)

assert (

testing.db.connect()

.exec_driver_sql("select count(1) from users")

.scalar()

== 1

)

sess.commit()

def test_autocommit_doesnt_raise_on_pending(self):

User, users = self.classes.User, self.tables.users

mapper(User, users)

session = create_session(autocommit=True)

session.add(User(name="ed"))

session.begin()

session.flush()

session.commit()

def test_active_flag(self):

sess = create_session(bind=config.db, autocommit=True)

assert not sess.is_active

sess.begin()

assert sess.is_active

sess.rollback()

assert not sess.is_active

@engines.close_open_connections

def test_add_delete(self):

User, Address, addresses, users = (

self.classes.User,

self.classes.Address,

self.tables.addresses,

self.tables.users,

)

s = create_session()

mapper(

User,

users,

properties={

"addresses": relationship(Address, cascade="all, delete")

},

)

mapper(Address, addresses)

user = User(name="u1")

assert_raises_message(

sa.exc.InvalidRequestError, "is not persisted", s.delete, user

)

s.add(user)

s.flush()

user = s.query(User).one()

s.expunge(user)

assert user not in s

# modify outside of session, assert changes remain/get saved

user.name = "fred"

s.add(user)

assert user in s

assert user in s.dirty

s.flush()

s.expunge_all()

assert s.query(User).count() == 1

user = s.query(User).one()

assert user.name == "fred"

# ensure its not dirty if no changes occur

s.expunge_all()

assert user not in s

s.add(user)

assert user in s

assert user not in s.dirty

s2 = create_session()

assert_raises_message(

sa.exc.InvalidRequestError,

"is already attached to session",

s2.delete,

user,

)

u2 = s2.query(User).get(user.id)

s2.expunge(u2)

assert_raises_message(

sa.exc.InvalidRequestError,

"another instance .* is already present",

s.delete,

u2,

)

s.expire(user)

s.expunge(user)

assert user not in s

s.delete(user)

assert user in s

s.flush()

assert user not in s

assert s.query(User).count() == 0

def test_already_attached(self):

User = self.classes.User

users = self.tables.users

mapper(User, users)

s1 = Session()

s2 = Session()

u1 = User(id=1, name="u1")

make_transient_to_detached(u1) # shorthand for actually persisting it

s1.add(u1)

assert_raises_message(

sa.exc.InvalidRequestError,

"Object '<User.*?>' is already attached to session",

s2.add,

u1,

)

assert u1 not in s2

assert not s2.identity_map.keys()

def test_identity_conflict(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

for s in (create_session(), create_session()):

users.delete().execute()

u1 = User(name="ed")

s.add(u1)

s.flush()

s.expunge(u1)

u2 = s.query(User).first()

s.expunge(u2)

s.identity_map.add(sa.orm.attributes.instance_state(u1))

assert_raises_message(

sa.exc.InvalidRequestError,

"Can't attach instance <User.*?>; another instance "

"with key .*? is already "

"present in this session.",

s.identity_map.add,

sa.orm.attributes.instance_state(u2),

)

def test_internal_identity_conflict_warning_weak(self):

# test for issue #4890

# see also test_naturalpks::ReversePKsTest::test_reverse

users, User = self.tables.users, self.classes.User

addresses, Address = self.tables.addresses, self.classes.Address

mapper(

User,

users,

properties={"addresses": relationship(Address, backref="user")},

)

mapper(Address, addresses)

session = Session()

@event.listens_for(session, "after_flush")

def load_collections(session, flush_context):

for target in set(session.new).union(session.dirty):

if isinstance(target, User):

target.addresses

u1 = User(name="u1")

a1 = Address(email_address="e1", user=u1)

session.add_all([u1, a1])

session.flush()

session.expire_all()

# create new Address via backref, so that u1.addresses remains

# expired and a2 is in pending mutations

a2 = Address(email_address="e2", user=u1)

assert "addresses" not in inspect(u1).dict

assert a2 in inspect(u1)._pending_mutations["addresses"].added_items

with assertions.expect_warnings(

r"Identity map already had an identity "

r"for \(.*Address.*\), replacing"

):

session.flush()

def test_pickled_update(self):

users, User = self.tables.users, pickleable.User

mapper(User, users)

sess1 = create_session()

sess2 = create_session()

u1 = User(name="u1")

sess1.add(u1)

assert_raises_message(

sa.exc.InvalidRequestError,

"already attached to session",

sess2.add,

u1,

)

u2 = pickle.loads(pickle.dumps(u1))

sess2.add(u2)

def test_duplicate_update(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

Session = sessionmaker()

sess = Session()

u1 = User(name="u1")

sess.add(u1)

sess.flush()

assert u1.id is not None

sess.expunge(u1)

assert u1 not in sess

assert Session.object_session(u1) is None

u2 = sess.query(User).get(u1.id)

assert u2 is not None and u2 is not u1

assert u2 in sess

assert_raises_message(

sa.exc.InvalidRequestError,

"Can't attach instance <User.*?>; another instance "

"with key .*? is already "

"present in this session.",

sess.add,

u1,

)

sess.expunge(u2)

assert u2 not in sess

assert Session.object_session(u2) is None

u1.name = "John"

u2.name = "Doe"

sess.add(u1)

assert u1 in sess

assert Session.object_session(u1) is sess

sess.flush()

sess.expunge_all()

u3 = sess.query(User).get(u1.id)

assert u3 is not u1 and u3 is not u2 and u3.name == u1.name

def test_no_double_save(self):

users = self.tables.users

sess = create_session()

class Foo(object):

def __init__(self):

sess.add(self)

class Bar(Foo):

def __init__(self):

sess.add(self)

Foo.__init__(self)

mapper(Foo, users)

mapper(Bar, users)

b = Bar()

assert b in sess

assert len(list(sess)) == 1

def test_identity_map_mutate(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

sess.add_all([User(name="u1"), User(name="u2"), User(name="u3")])

sess.commit()

# TODO: what are we testing here ? that iteritems() can

# withstand a change? should this be

# more directly attempting to manipulate the identity_map ?

u1, u2, u3 = sess.query(User).all()

for i, (key, value) in enumerate(iter(sess.identity_map.items())):

if i == 2:

del u3

gc_collect()

def _test_extra_dirty_state(self):

users, User = self.tables.users, self.classes.User

m = mapper(User, users)

s = Session()

@event.listens_for(m, "after_update")

def e(mapper, conn, target):

sess = object_session(target)

for entry in list(sess.identity_map.values()):

entry.name = "5"

a1, a2 = User(name="1"), User(name="2")

s.add_all([a1, a2])

s.commit()

a1.name = "3"

return s, a1, a2

def test_extra_dirty_state_post_flush_warning(self):

s, a1, a2 = self._test_extra_dirty_state()

assert_raises_message(

sa.exc.SAWarning,

"Attribute history events accumulated on 1 previously "

"clean instances",

s.commit,

)

def test_extra_dirty_state_post_flush_state(self):

s, a1, a2 = self._test_extra_dirty_state()

canary = []

@event.listens_for(s, "after_flush_postexec")

def e(sess, ctx):

canary.append(bool(sess.identity_map._modified))

@testing.emits_warning("Attribute")

def go():

s.commit()

go()

eq_(canary, [False])

def test_deleted_auto_expunged(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

sess.add(User(name="x"))

sess.commit()

u1 = sess.query(User).first()

sess.delete(u1)

assert not was_deleted(u1)

sess.flush()

assert was_deleted(u1)

assert u1 not in sess

assert object_session(u1) is sess

sess.commit()

assert object_session(u1) is None

def test_explicit_expunge_pending(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

u1 = User(name="x")

sess.add(u1)

sess.flush()

sess.expunge(u1)

assert u1 not in sess

assert object_session(u1) is None

sess.rollback()

assert u1 not in sess

assert object_session(u1) is None

def test_explicit_expunge_deleted(self):

users, User = self.tables.users, self.classes.User

mapper(User, users)

sess = Session()

sess.add(User(name="x"))

sess.commit()

u1 = sess.query(User).first()

sess.delete(u1)

sess.flush()

assert was_deleted(u1)

assert u1 not in sess

assert object_session(u1) is sess

sess.expunge(u1)