import collections

from contextlib import contextmanager

import csv

import operator

from sqlalchemy import CHAR

from sqlalchemy import column

from sqlalchemy import exc

from sqlalchemy import exc as sa_exc

from sqlalchemy import ForeignKey

from sqlalchemy import func

from sqlalchemy import INT

from sqlalchemy import Integer

from sqlalchemy import literal

from sqlalchemy import literal_column

from sqlalchemy import MetaData

from sqlalchemy import select

from sqlalchemy import sql

from sqlalchemy import String

from sqlalchemy import table

from sqlalchemy import testing

from sqlalchemy import text

from sqlalchemy import true

from sqlalchemy import type_coerce

from sqlalchemy import TypeDecorator

from sqlalchemy import util

from sqlalchemy import VARCHAR

from sqlalchemy.engine import cursor as _cursor

from sqlalchemy.engine import default

from sqlalchemy.engine import Row

from sqlalchemy.ext.compiler import compiles

from sqlalchemy.future import select as future_select

from sqlalchemy.sql import ColumnElement

from sqlalchemy.sql import expression

from sqlalchemy.sql.selectable import TextualSelect

from sqlalchemy.sql.sqltypes import NULLTYPE

from sqlalchemy.sql.util import ClauseAdapter

from sqlalchemy.testing import assert_raises

from sqlalchemy.testing import assert_raises_message

from sqlalchemy.testing import assertions

from sqlalchemy.testing import engines

from sqlalchemy.testing import eq_

from sqlalchemy.testing import fixtures

from sqlalchemy.testing import in_

from sqlalchemy.testing import is_

from sqlalchemy.testing import is_true

from sqlalchemy.testing import le_

from sqlalchemy.testing import ne_

from sqlalchemy.testing import not_in_

from sqlalchemy.testing.mock import Mock

from sqlalchemy.testing.mock import patch

from sqlalchemy.testing.schema import Column

from sqlalchemy.testing.schema import Table

from sqlalchemy.util import collections_abc

class CursorResultTest(fixtures.TablesTest):

__backend__ = True

@classmethod

def define_tables(cls, metadata):

Table(

"users",

metadata,

Column(

"user_id", INT, primary_key=True, test_needs_autoincrement=True

),

Column("user_name", VARCHAR(20)),

test_needs_acid=True,

)

Table(

"addresses",

metadata,

Column(

"address_id",

Integer,

primary_key=True,

test_needs_autoincrement=True,

),

Column("user_id", Integer, ForeignKey("users.user_id")),

Column("address", String(30)),

test_needs_acid=True,

)

Table(

"users2",

metadata,

Column("user_id", INT, primary_key=True),

Column("user_name", VARCHAR(20)),

test_needs_acid=True,

)

def test_row_iteration(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

{"user_id": 7, "user_name": "jack"},

{"user_id": 8, "user_name": "ed"},

{"user_id": 9, "user_name": "fred"},

)

r = connection.execute(users.select())

rows = []

for row in r:

rows.append(row)

eq_(len(rows), 3)

def test_row_next(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

{"user_id": 7, "user_name": "jack"},

{"user_id": 8, "user_name": "ed"},

{"user_id": 9, "user_name": "fred"},

)

r = connection.execute(users.select())

rows = []

while True:

row = next(r, "foo")

if row == "foo":

break

rows.append(row)

eq_(len(rows), 3)

@testing.requires.subqueries

def test_anonymous_rows(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

{"user_id": 7, "user_name": "jack"},

{"user_id": 8, "user_name": "ed"},

{"user_id": 9, "user_name": "fred"},

)

sel = (

select([users.c.user_id])

.where(users.c.user_name == "jack")

.scalar_subquery()

)

for row in connection.execute(

select([sel + 1, sel + 3], bind=users.bind)

):

eq_(row._mapping["anon_1"], 8)

eq_(row._mapping["anon_2"], 10)

def test_row_comparison(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=7, user_name="jack")

rp = connection.execute(users.select()).first()

eq_(rp, rp)

is_(not (rp != rp), True)

equal = (7, "jack")

eq_(rp, equal)

eq_(equal, rp)

is_((not (rp != equal)), True)

is_(not (equal != equal), True)

def endless():

while True:

yield 1

ne_(rp, endless())

ne_(endless(), rp)

# test that everything compares the same

# as it would against a tuple

for compare in [False, 8, endless(), "xyz", (7, "jack")]:

for op in [

operator.eq,

operator.ne,

operator.gt,

operator.lt,

operator.ge,

operator.le,

]:

try:

control = op(equal, compare)

except TypeError:

# Py3K raises TypeError for some invalid comparisons

assert_raises(TypeError, op, rp, compare)

else:

eq_(control, op(rp, compare))

try:

control = op(compare, equal)

except TypeError:

# Py3K raises TypeError for some invalid comparisons

assert_raises(TypeError, op, compare, rp)

else:

eq_(control, op(compare, rp))

@testing.provide_metadata

def test_column_label_overlap_fallback(self, connection):

content = Table("content", self.metadata, Column("type", String(30)))

bar = Table("bar", self.metadata, Column("content_type", String(30)))

self.metadata.create_all(connection)

connection.execute(content.insert().values(type="t1"))

row = connection.execute(content.select(use_labels=True)).first()

in_(content.c.type, row._mapping)

not_in_(bar.c.content_type, row._mapping)

not_in_(bar.c.content_type, row._mapping)

row = connection.execute(

select([func.now().label("content_type")])

).first()

not_in_(content.c.type, row._mapping)

not_in_(bar.c.content_type, row._mapping)

def test_pickled_rows(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

{"user_id": 7, "user_name": "jack"},

{"user_id": 8, "user_name": "ed"},

{"user_id": 9, "user_name": "fred"},

)

for pickle in False, True:

for use_labels in False, True:

result = connection.execute(

users.select(use_labels=use_labels).order_by(

users.c.user_id

)

).fetchall()

if pickle:

result = util.pickle.loads(util.pickle.dumps(result))

eq_(result, [(7, "jack"), (8, "ed"), (9, "fred")])

if use_labels:

eq_(result[0]._mapping["users_user_id"], 7)

eq_(

list(result[0]._fields),

["users_user_id", "users_user_name"],

)

else:

eq_(result[0]._mapping["user_id"], 7)

eq_(list(result[0]._fields), ["user_id", "user_name"])

eq_(result[0][0], 7)

assert_raises(

exc.NoSuchColumnError, lambda: result[0]["fake key"]

)

assert_raises(

exc.NoSuchColumnError,

lambda: result[0]._mapping["fake key"],

)

def test_column_error_printing(self, connection):

result = connection.execute(select([1]))

row = result.first()

class unprintable(object):

def __str__(self):

raise ValueError("nope")

msg = r"Could not locate column in row for column '%s'"

for accessor, repl in [

("x", "x"),

(Column("q", Integer), "q"),

(Column("q", Integer) + 12, r"q \+ :q_1"),

(unprintable(), "unprintable element.*"),

]:

assert_raises_message(

exc.NoSuchColumnError, msg % repl, result._getter, accessor

)

is_(result._getter(accessor, False), None)

assert_raises_message(

exc.NoSuchColumnError,

msg % repl,

lambda: row._mapping[accessor],

)

def test_fetchmany(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=7, user_name="jack")

connection.execute(users.insert(), user_id=8, user_name="ed")

connection.execute(users.insert(), user_id=9, user_name="fred")

r = connection.execute(users.select())

rows = []

for row in r.fetchmany(size=2):

rows.append(row)

eq_(len(rows), 2)

def test_column_slices(self, connection):

users = self.tables.users

addresses = self.tables.addresses

connection.execute(users.insert(), user_id=1, user_name="john")

connection.execute(users.insert(), user_id=2, user_name="jack")

connection.execute(

addresses.insert(), address_id=1, user_id=2, address="foo@bar.com"

)

r = connection.execute(

text("select * from addresses", bind=testing.db)

).first()

eq_(r[0:1], (1,))

eq_(r[1:], (2, "foo@bar.com"))

eq_(r[:-1], (1, 2))

def test_column_accessor_basic_compiled_mapping(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

dict(user_id=1, user_name="john"),

dict(user_id=2, user_name="jack"),

)

r = connection.execute(users.select(users.c.user_id == 2)).first()

eq_(r.user_id, 2)

eq_(r._mapping["user_id"], 2)

eq_(r._mapping[users.c.user_id], 2)

eq_(r.user_name, "jack")

eq_(r._mapping["user_name"], "jack")

eq_(r._mapping[users.c.user_name], "jack")

def test_column_accessor_basic_compiled_traditional(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

dict(user_id=1, user_name="john"),

dict(user_id=2, user_name="jack"),

)

r = connection.execute(users.select(users.c.user_id == 2)).first()

eq_(r.user_id, 2)

eq_(r._mapping["user_id"], 2)

eq_(r._mapping[users.c.user_id], 2)

eq_(r.user_name, "jack")

eq_(r._mapping["user_name"], "jack")

eq_(r._mapping[users.c.user_name], "jack")

def test_row_getitem_string(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

dict(user_id=1, user_name="john"),

dict(user_id=2, user_name="jack"),

)

r = connection.execute(

text("select * from users where user_id=2")

).first()

eq_(r._mapping["user_name"], "jack")

def test_column_accessor_basic_text(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

dict(user_id=1, user_name="john"),

dict(user_id=2, user_name="jack"),

)

r = connection.execute(

text("select * from users where user_id=2")

).first()

eq_(r.user_id, 2)

eq_(r.user_name, "jack")

eq_(r._mapping["user_id"], 2)

eq_(r.user_name, "jack")

eq_(r._mapping["user_name"], "jack")

def test_column_accessor_text_colexplicit(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

dict(user_id=1, user_name="john"),

dict(user_id=2, user_name="jack"),

)

r = connection.execute(

text("select * from users where user_id=2").columns(

users.c.user_id, users.c.user_name

)

).first()

eq_(r.user_id, 2)

eq_(r._mapping["user_id"], 2)

eq_(r._mapping[users.c.user_id], 2)

eq_(r.user_name, "jack")

eq_(r._mapping["user_name"], "jack")

eq_(r._mapping[users.c.user_name], "jack")

def test_column_accessor_textual_select(self, connection):

users = self.tables.users

connection.execute(

users.insert(),

dict(user_id=1, user_name="john"),

dict(user_id=2, user_name="jack"),

)

# this will create column() objects inside

# the select(), these need to match on name anyway

r = connection.execute(

select([column("user_id"), column("user_name")])

.select_from(table("users"))

.where(text("user_id=2"))

).first()

# keyed access works in many ways

eq_(r.user_id, 2)

eq_(r.user_name, "jack")

eq_(r._mapping["user_id"], 2)

eq_(r.user_name, "jack")

eq_(r._mapping["user_name"], "jack")

def test_column_accessor_dotted_union(self, connection):

users = self.tables.users

connection.execute(users.insert(), dict(user_id=1, user_name="john"))

# test a little sqlite < 3.10.0 weirdness - with the UNION,

# cols come back as "users.user_id" in cursor.description

r = connection.execute(

text(

"select users.user_id, users.user_name "

"from users "

"UNION select users.user_id, "

"users.user_name from users"

)

).first()

eq_(r._mapping["user_id"], 1)

eq_(r._mapping["user_name"], "john")

eq_(list(r._fields), ["user_id", "user_name"])

def test_column_accessor_sqlite_raw(self, connection):

users = self.tables.users

connection.execute(users.insert(), dict(user_id=1, user_name="john"))

r = connection.execute(

text(

"select users.user_id, users.user_name "

"from users "

"UNION select users.user_id, "

"users.user_name from users",

bind=testing.db,

).execution_options(sqlite_raw_colnames=True)

).first()

if testing.against("sqlite < 3.10.0"):

not_in_("user_id", r)

not_in_("user_name", r)

eq_(r["users.user_id"], 1)

eq_(r["users.user_name"], "john")

eq_(list(r._fields), ["users.user_id", "users.user_name"])

else:

not_in_("users.user_id", r._mapping)

not_in_("users.user_name", r._mapping)

eq_(r._mapping["user_id"], 1)

eq_(r._mapping["user_name"], "john")

eq_(list(r._fields), ["user_id", "user_name"])

def test_column_accessor_sqlite_translated(self, connection):

users = self.tables.users

connection.execute(users.insert(), dict(user_id=1, user_name="john"))

r = connection.execute(

text(

"select users.user_id, users.user_name "

"from users "

"UNION select users.user_id, "

"users.user_name from users",

bind=testing.db,

)

).first()

eq_(r._mapping["user_id"], 1)

eq_(r._mapping["user_name"], "john")

if testing.against("sqlite < 3.10.0"):

eq_(r._mapping["users.user_id"], 1)

eq_(r._mapping["users.user_name"], "john")

else:

not_in_("users.user_id", r._mapping)

not_in_("users.user_name", r._mapping)

eq_(list(r._fields), ["user_id", "user_name"])

def test_column_accessor_labels_w_dots(self, connection):

users = self.tables.users

connection.execute(users.insert(), dict(user_id=1, user_name="john"))

# test using literal tablename.colname

r = connection.execute(

text(

'select users.user_id AS "users.user_id", '

'users.user_name AS "users.user_name" '

"from users",

bind=testing.db,

).execution_options(sqlite_raw_colnames=True)

).first()

eq_(r._mapping["users.user_id"], 1)

eq_(r._mapping["users.user_name"], "john")

not_in_("user_name", r._mapping)

eq_(list(r._fields), ["users.user_id", "users.user_name"])

def test_column_accessor_unary(self, connection):

users = self.tables.users

connection.execute(users.insert(), dict(user_id=1, user_name="john"))

# unary expressions

r = connection.execute(

select([users.c.user_name.distinct()]).order_by(users.c.user_name)

).first()

eq_(r._mapping[users.c.user_name], "john")

eq_(r.user_name, "john")

def test_column_accessor_err(self, connection):

r = connection.execute(select([1])).first()

assert_raises_message(

AttributeError,

"Could not locate column in row for column 'foo'",

getattr,

r,

"foo",

)

assert_raises_message(

KeyError,

"Could not locate column in row for column 'foo'",

lambda: r._mapping["foo"],

)

def test_graceful_fetch_on_non_rows(self):

"""test that calling fetchone() etc. on a result that doesn't

return rows fails gracefully.

"""

# these proxies don't work with no cursor.description present.

# so they don't apply to this test at the moment.

# result.FullyBufferedCursorResult,

# result.BufferedRowCursorResult,

# result.BufferedColumnCursorResult

users = self.tables.users

conn = testing.db.connect()

for meth in [

lambda r: r.fetchone(),

lambda r: r.fetchall(),

lambda r: r.first(),

lambda r: r.scalar(),

lambda r: r.fetchmany(),

lambda r: r._getter("user"),

lambda r: r.keys(),

lambda r: r.columns("user"),

lambda r: r.cursor_strategy.fetchone(r),

]:

trans = conn.begin()

result = conn.execute(users.insert(), user_id=1)

assert_raises_message(

exc.ResourceClosedError,

"This result object does not return rows. "

"It has been closed automatically.",

meth,

result,

)

trans.rollback()

def test_fetchone_til_end(self, connection):

result = connection.exec_driver_sql("select * from users")

eq_(result.fetchone(), None)

eq_(result.fetchone(), None)

eq_(result.fetchone(), None)

result.close()

assert_raises_message(

exc.ResourceClosedError,

"This result object is closed.",

result.fetchone,

)

def test_connectionless_autoclose_rows_exhausted(self):

# TODO: deprecate for 2.0

users = self.tables.users

with testing.db.connect() as conn:

conn.execute(users.insert(), dict(user_id=1, user_name="john"))

result = testing.db.execute(text("select * from users"))

connection = result.connection

assert not connection.closed

eq_(result.fetchone(), (1, "john"))

assert not connection.closed

eq_(result.fetchone(), None)

assert connection.closed

@testing.requires.returning

def test_connectionless_autoclose_crud_rows_exhausted(self):

# TODO: deprecate for 2.0

users = self.tables.users

stmt = (

users.insert()

.values(user_id=1, user_name="john")

.returning(users.c.user_id)

)

result = testing.db.execute(stmt)

connection = result.connection

assert not connection.closed

eq_(result.fetchone(), (1,))

assert not connection.closed

eq_(result.fetchone(), None)

assert connection.closed

def test_connectionless_autoclose_no_rows(self):

# TODO: deprecate for 2.0

result = testing.db.execute(text("select * from users"))

connection = result.connection

assert not connection.closed

eq_(result.fetchone(), None)

assert connection.closed

@testing.requires.updateable_autoincrement_pks

def test_connectionless_autoclose_no_metadata(self):

# TODO: deprecate for 2.0

result = testing.db.execute(text("update users set user_id=5"))

connection = result.connection

assert connection.closed

assert_raises_message(

exc.ResourceClosedError,

"This result object does not return rows.",

result.fetchone,

)

assert_raises_message(

exc.ResourceClosedError,

"This result object does not return rows.",

result.keys,

)

def test_row_case_sensitive(self, connection):

row = connection.execute(

select(

[

literal_column("1").label("case_insensitive"),

literal_column("2").label("CaseSensitive"),

]

)

).first()

eq_(list(row._fields), ["case_insensitive", "CaseSensitive"])

in_("case_insensitive", row._keymap)

in_("CaseSensitive", row._keymap)

not_in_("casesensitive", row._keymap)

eq_(row._mapping["case_insensitive"], 1)

eq_(row._mapping["CaseSensitive"], 2)

assert_raises(KeyError, lambda: row._mapping["Case_insensitive"])

assert_raises(KeyError, lambda: row._mapping["casesensitive"])

def test_row_case_sensitive_unoptimized(self):

with engines.testing_engine().connect() as ins_conn:

row = ins_conn.execute(

select(

[

literal_column("1").label("case_insensitive"),

literal_column("2").label("CaseSensitive"),

text("3 AS screw_up_the_cols"),

]

)

).first()

eq_(

list(row._fields),

["case_insensitive", "CaseSensitive", "screw_up_the_cols"],

)

in_("case_insensitive", row._keymap)

in_("CaseSensitive", row._keymap)

not_in_("casesensitive", row._keymap)

eq_(row._mapping["case_insensitive"], 1)

eq_(row._mapping["CaseSensitive"], 2)

eq_(row._mapping["screw_up_the_cols"], 3)

assert_raises(KeyError, lambda: row._mapping["Case_insensitive"])

assert_raises(KeyError, lambda: row._mapping["casesensitive"])

assert_raises(KeyError, lambda: row._mapping["screw_UP_the_cols"])

def test_row_as_args(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=1, user_name="john")

r = connection.execute(users.select(users.c.user_id == 1)).first()

connection.execute(users.delete())

connection.execute(users.insert(), r._mapping)

eq_(connection.execute(users.select()).fetchall(), [(1, "john")])

def test_result_as_args(self, connection):

users = self.tables.users

users2 = self.tables.users2

connection.execute(

users.insert(),

[

dict(user_id=1, user_name="john"),

dict(user_id=2, user_name="ed"),

],

)

r = connection.execute(users.select())

connection.execute(users2.insert(), [row._mapping for row in r])

eq_(

connection.execute(

users2.select().order_by(users2.c.user_id)

).fetchall(),

[(1, "john"), (2, "ed")],

)

connection.execute(users2.delete())

r = connection.execute(users.select())

connection.execute(users2.insert(), *[row._mapping for row in r])

eq_(

connection.execute(

users2.select().order_by(users2.c.user_id)

).fetchall(),

[(1, "john"), (2, "ed")],

)

@testing.requires.duplicate_names_in_cursor_description

def test_ambiguous_column(self, connection):

users = self.tables.users

addresses = self.tables.addresses

connection.execute(users.insert(), user_id=1, user_name="john")

result = connection.execute(users.outerjoin(addresses).select())

r = result.first()

assert_raises_message(

exc.InvalidRequestError,

"Ambiguous column name",

lambda: r._mapping["user_id"],

)

assert_raises_message(

exc.InvalidRequestError,

"Ambiguous column name",

result._getter,

"user_id",

)

# pure positional targeting; users.c.user_id

# and addresses.c.user_id are known!

# works as of 1.1 issue #3501

eq_(r._mapping[users.c.user_id], 1)

eq_(r._mapping[addresses.c.user_id], None)

# try to trick it - fake_table isn't in the result!

# we get the correct error

fake_table = Table("fake", MetaData(), Column("user_id", Integer))

assert_raises_message(

exc.InvalidRequestError,

"Could not locate column in row for column 'fake.user_id'",

lambda: r._mapping[fake_table.c.user_id],

)

r = util.pickle.loads(util.pickle.dumps(r))

assert_raises_message(

exc.InvalidRequestError,

"Ambiguous column name",

lambda: r._mapping["user_id"],

)

@testing.requires.duplicate_names_in_cursor_description

def test_ambiguous_column_by_col(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=1, user_name="john")

ua = users.alias()

u2 = users.alias()

result = connection.execute(

select([users.c.user_id, ua.c.user_id]).select_from(

users.join(ua, true())

)

)

row = result.first()

# as of 1.1 issue #3501, we use pure positional

# targeting for the column objects here

eq_(row._mapping[users.c.user_id], 1)

eq_(row._mapping[ua.c.user_id], 1)

# this now works as of 1.1 issue #3501;

# previously this was stuck on "ambiguous column name"

assert_raises_message(

exc.InvalidRequestError,

"Could not locate column in row",

lambda: row._mapping[u2.c.user_id],

)

@testing.requires.duplicate_names_in_cursor_description

def test_ambiguous_column_contains(self, connection):

users = self.tables.users

addresses = self.tables.addresses

# ticket 2702. in 0.7 we'd get True, False.

# in 0.8, both columns are present so it's True;

# but when they're fetched you'll get the ambiguous error.

connection.execute(users.insert(), user_id=1, user_name="john")

result = connection.execute(

select([users.c.user_id, addresses.c.user_id]).select_from(

users.outerjoin(addresses)

)

)

row = result.first()

eq_(

set(

[

users.c.user_id in row._mapping,

addresses.c.user_id in row._mapping,

]

),

set([True]),

)

def test_loose_matching_one(self, connection):

users = self.tables.users

addresses = self.tables.addresses

connection.execute(users.insert(), {"user_id": 1, "user_name": "john"})

connection.execute(

addresses.insert(),

{"address_id": 1, "user_id": 1, "address": "email"},

)

# use some column labels in the SELECT

result = connection.execute(

TextualSelect(

text(

"select users.user_name AS users_user_name, "

"users.user_id AS user_id, "

"addresses.address_id AS address_id "

"FROM users JOIN addresses "

"ON users.user_id = addresses.user_id "

"WHERE users.user_id=1 "

),

[users.c.user_id, users.c.user_name, addresses.c.address_id],

positional=False,

)

)

row = result.first()

eq_(row._mapping[users.c.user_id], 1)

eq_(row._mapping[users.c.user_name], "john")

def test_loose_matching_two(self, connection):

users = self.tables.users

addresses = self.tables.addresses

connection.execute(users.insert(), {"user_id": 1, "user_name": "john"})

connection.execute(

addresses.insert(),

{"address_id": 1, "user_id": 1, "address": "email"},

)

# use some column labels in the SELECT

result = connection.execute(

TextualSelect(

text(

"select users.user_name AS users_user_name, "

"users.user_id AS user_id, "

"addresses.user_id "

"FROM users JOIN addresses "

"ON users.user_id = addresses.user_id "

"WHERE users.user_id=1 "

),

[users.c.user_id, users.c.user_name, addresses.c.user_id],

positional=False,

)

)

row = result.first()

assert_raises_message(

exc.InvalidRequestError,

"Ambiguous column name",

lambda: row._mapping[users.c.user_id],

)

assert_raises_message(

exc.InvalidRequestError,

"Ambiguous column name",

lambda: row._mapping[addresses.c.user_id],

)

eq_(row._mapping[users.c.user_name], "john")

def test_ambiguous_column_by_col_plus_label(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=1, user_name="john")

result = connection.execute(

select(

[

users.c.user_id,

type_coerce(users.c.user_id, Integer).label("foo"),

]

)

)

row = result.first()

eq_(row._mapping[users.c.user_id], 1)

eq_(row[1], 1)

def test_fetch_partial_result_map(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=7, user_name="ed")

t = text("select * from users").columns(user_name=String())

eq_(connection.execute(t).fetchall(), [(7, "ed")])

def test_fetch_unordered_result_map(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=7, user_name="ed")

class Goofy1(TypeDecorator):

impl = String

def process_result_value(self, value, dialect):

return value + "a"

class Goofy2(TypeDecorator):

impl = String

def process_result_value(self, value, dialect):

return value + "b"

class Goofy3(TypeDecorator):

impl = String

def process_result_value(self, value, dialect):

return value + "c"

t = text(

"select user_name as a, user_name as b, "

"user_name as c from users"

).columns(a=Goofy1(), b=Goofy2(), c=Goofy3())

eq_(connection.execute(t).fetchall(), [("eda", "edb", "edc")])

@testing.requires.subqueries

def test_column_label_targeting(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=7, user_name="ed")

for s in (

users.select().alias("foo"),

users.select().alias(users.name),

):

row = connection.execute(s.select(use_labels=True)).first()

eq_(row._mapping[s.c.user_id], 7)

eq_(row._mapping[s.c.user_name], "ed")

@testing.requires.python3

def test_ro_mapping_py3k(self, connection):

users = self.tables.users

connection.execute(users.insert(), user_id=1, user_name="foo")

result = connection.execute(users.select())

row = result.first()

dict_row = row._asdict()

# dictionaries aren't ordered in Python 3 until 3.7

odict_row = collections.OrderedDict(

[("user_id", 1), ("user_name", "foo")]

)