if col.autoincrement and \
issubclass(col.type._type_affinity, types.Integer) and \
not col.foreign_keys and \
- isinstance(col.default, (type(None), Sequence)):
- # don't look at server_default here since different backends may
- # or may not have a server_default, e.g. postgresql reflected
- # SERIAL cols will have a DefaultClause here but are still
- # autoincrement.
+ isinstance(col.default, (type(None), Sequence)) and \
+ (col.server_default is None or col.server_default.reflected):
return col
@property
__visit_name__ = 'default_generator'
is_sequence = False
+ is_server_default = False
def __init__(self, for_update=False):
self.for_update = for_update
INSERT.
"""
+ is_server_default = True
+ reflected = False
def __init__(self, for_update=False):
self.for_update = for_update
"""
- def __init__(self, arg, for_update=False):
+ def __init__(self, arg, for_update=False, _reflected=False):
util.assert_arg_type(arg, (basestring,
expression.ClauseElement,
expression._TextClause), 'arg')
super(DefaultClause, self).__init__(for_update)
self.arg = arg
+ self.reflected = _reflected
def __repr__(self):
return "DefaultClause(%r, for_update=%r)" % \
self._bind = bind
bind = property(bind, _set_bind)
-class _UpdateBase(Executable, ClauseElement):
+class UpdateBase(Executable, ClauseElement):
"""Form the base for ``INSERT``, ``UPDATE``, and ``DELETE`` statements."""
__visit_name__ = 'update_base'
"""
self._returning = cols
-class _ValuesBase(_UpdateBase):
+class ValuesBase(UpdateBase):
+ """Supplies support for :meth:`.ValuesBase.values` to INSERT and UPDATE constructs."""
__visit_name__ = 'values_base'
self.parameters.update(self._process_colparams(v))
self.parameters.update(kwargs)
-class Insert(_ValuesBase):
+class Insert(ValuesBase):
"""Represent an INSERT construct.
The :class:`Insert` object is created using the :func:`insert()` function.
prefixes=None,
returning=None,
**kwargs):
- _ValuesBase.__init__(self, table, values)
+ ValuesBase.__init__(self, table, values)
self._bind = bind
self.select = None
self.inline = inline
clause = _literal_as_text(clause)
self._prefixes = self._prefixes + (clause,)
-class Update(_ValuesBase):
+class Update(ValuesBase):
"""Represent an Update construct.
The :class:`Update` object is created using the :func:`update()` function.
bind=None,
returning=None,
**kwargs):
- _ValuesBase.__init__(self, table, values)
+ ValuesBase.__init__(self, table, values)
self._bind = bind
self._returning = returning
if whereclause is not None:
self._whereclause = _literal_as_text(whereclause)
-class Delete(_UpdateBase):
+class Delete(UpdateBase):
"""Represent a DELETE construct.
The :class:`Delete` object is created using the :func:`delete()` function.
import sqlalchemy as sa
from test.lib import testing, engines
from sqlalchemy import MetaData, Integer, String, ForeignKey, Boolean, exc,\
- Sequence, Column, func, literal
+ Sequence, func, literal
from sqlalchemy.types import TypeDecorator
-from test.lib.schema import Table
+from test.lib.schema import Table, Column
from test.lib.testing import eq_
from test.sql import _base
class MyInteger(TypeDecorator):
impl = Integer
def process_bind_param(self, value, dialect):
+ if value is None:
+ return None
return int(value[4:])
def process_result_value(self, value, dialect):
+ if value is None:
+ return None
return "INT_%d" % value
cls.MyInteger = MyInteger
def _run_test(self, *arg, **kw):
implicit_returning = kw.pop('implicit_returning', True)
kw['primary_key'] = True
+ if kw.get('autoincrement', True):
+ kw['test_needs_autoincrement'] = True
t = Table('x', metadata,
Column('y', self.MyInteger, *arg, **kw),
Column('data', Integer),
t.create()
r = t.insert().values(data=5).execute()
- eq_(r.inserted_primary_key, ['INT_1'])
+
+ # we don't pre-fetch 'server_default'.
+ if 'server_default' in kw and (not testing.db.dialect.implicit_returning or not implicit_returning):
+ eq_(r.inserted_primary_key, [None])
+ else:
+ eq_(r.inserted_primary_key, ['INT_1'])
r.close()
eq_(
def test_sequence(self):
self._run_test(Sequence('foo_seq'))
- @testing.fails_on('mysql', "Pending [ticket:2021]")
def test_server_default(self):
- # note that the MySQL dialect has to not render AUTOINCREMENT on this one
self._run_test(server_default='1',)
- @testing.fails_on('mysql', "Pending [ticket:2021]")
- @testing.fails_on('sqlite', "Pending [ticket:2021]")
def test_server_default_no_autoincrement(self):
self._run_test(server_default='1', autoincrement=False)
def test_server_default_no_implicit_returning(self):
self._run_test(server_default='1', autoincrement=False)
+class ServerDefaultsOnPKTest(testing.TestBase):
+ @testing.provide_metadata
+ def test_string_default_none_on_insert(self):
+ """Test that without implicit returning, we return None for
+ a string server default.
+
+ That is, we don't want to attempt to pre-execute "server_default"
+ generically - the user should use a Python side-default for a case
+ like this. Testing that all backends do the same thing here.
+
+ """
+ t = Table('x', metadata,
+ Column('y', String(10), server_default='key_one', primary_key=True),
+ Column('data', String(10)),
+ implicit_returning=False
+ )
+ metadata.create_all()
+ r = t.insert().execute(data='data')
+ eq_(r.inserted_primary_key, [None])
+ eq_(
+ t.select().execute().fetchall(),
+ [('key_one', 'data')]
+ )
+
+ @testing.requires.returning
+ @testing.provide_metadata
+ def test_string_default_on_insert_with_returning(self):
+ """With implicit_returning, we get a string PK default back no problem."""
+ t = Table('x', metadata,
+ Column('y', String(10), server_default='key_one', primary_key=True),
+ Column('data', String(10))
+ )
+ metadata.create_all()
+ r = t.insert().execute(data='data')
+ eq_(r.inserted_primary_key, ['key_one'])
+ eq_(
+ t.select().execute().fetchall(),
+ [('key_one', 'data')]
+ )
+
+ @testing.provide_metadata
+ def test_int_default_none_on_insert(self):
+ t = Table('x', metadata,
+ Column('y', Integer,
+ server_default='5', primary_key=True),
+ Column('data', String(10)),
+ implicit_returning=False
+ )
+ assert t._autoincrement_column is None
+ metadata.create_all()
+ r = t.insert().execute(data='data')
+ eq_(r.inserted_primary_key, [None])
+ if testing.against('sqlite'):
+ eq_(
+ t.select().execute().fetchall(),
+ [(1, 'data')]
+ )
+ else:
+ eq_(
+ t.select().execute().fetchall(),
+ [(5, 'data')]
+ )
+ @testing.fails_on('firebird', "col comes back as autoincrement")
+ @testing.fails_on('sqlite', "col comes back as autoincrement")
+ @testing.fails_on('oracle', "col comes back as autoincrement")
+ @testing.provide_metadata
+ def test_autoincrement_reflected_from_server_default(self):
+ t = Table('x', metadata,
+ Column('y', Integer,
+ server_default='5', primary_key=True),
+ Column('data', String(10)),
+ implicit_returning=False
+ )
+ assert t._autoincrement_column is None
+ metadata.create_all()
+
+ m2 = MetaData(metadata.bind)
+ t2 = Table('x', m2, autoload=True, implicit_returning=False)
+ assert t2._autoincrement_column is None
+
+ @testing.fails_on('firebird', "attempts to insert None")
+ @testing.fails_on('sqlite', "returns a value")
+ @testing.provide_metadata
+ def test_int_default_none_on_insert_reflected(self):
+ t = Table('x', metadata,
+ Column('y', Integer,
+ server_default='5', primary_key=True),
+ Column('data', String(10)),
+ implicit_returning=False
+ )
+ metadata.create_all()
+
+ m2 = MetaData(metadata.bind)
+ t2 = Table('x', m2, autoload=True, implicit_returning=False)
+
+ r = t2.insert().execute(data='data')
+ eq_(r.inserted_primary_key, [None])
+ if testing.against('sqlite'):
+ eq_(
+ t2.select().execute().fetchall(),
+ [(1, 'data')]
+ )
+ else:
+ eq_(
+ t2.select().execute().fetchall(),
+ [(5, 'data')]
+ )
+
+ @testing.requires.returning
+ @testing.provide_metadata
+ def test_int_default_on_insert_with_returning(self):
+ t = Table('x', metadata,
+ Column('y', Integer,
+ server_default='5', primary_key=True),
+ Column('data', String(10))
+ )
+
+ metadata.create_all()
+ r = t.insert().execute(data='data')
+ eq_(r.inserted_primary_key, [5])
+ eq_(
+ t.select().execute().fetchall(),
+ [(5, 'data')]
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
