- Added a generic BigInteger type, compiles to
BIGINT or NUMBER(19). [ticket:1125]
-
+
+-ext
+ - sqlsoup has been overhauled to explicitly support an 0.5 style
+ session, using autocommit=False, autoflush=True. Default
+ behavior of SQLSoup now requires the usual usage of commit()
+ and rollback(), which have been added to its interface. An
+ explcit Session or scoped_session can be passed to the
+ constructor, allowing these arguments to be overridden.
+
+ - sqlsoup db.<sometable>.update() and delete() now call
+ query(cls).update() and delete(), respectively.
+
+ - sqlsoup now has execute() and connection(), which call upon
+ the Session methods of those names, ensuring that the bind is
+ in terms of the SqlSoup object's bind.
+
0.5.7
=====
- orm
['ed', 'fakeuser']
{stop}[<User('ed','Ed Jones', 'f8s7ccs')>]
+.. _ormtutorial_querying:
+
Querying
========
SqlSoup
=======
-:author: Jonathan Ellis
-
-SqlSoup creates mapped classes on the fly from tables, which are automatically reflected from the database based on name. It is essentially a nicer version of the "row data gateway" pattern.
-
-.. sourcecode:: python+sql
-
- >>> from sqlalchemy.ext.sqlsoup import SqlSoup
- >>> soup = SqlSoup('sqlite:///')
-
- >>> db.users.select(order_by=[db.users.c.name])
- [MappedUsers(name='Bhargan Basepair',email='basepair@example.edu',password='basepair',classname=None,admin=1),
- MappedUsers(name='Joe Student',email='student@example.edu',password='student',classname=None,admin=0)]
-
-Full SqlSoup documentation is on the `SQLAlchemy Wiki <http://www.sqlalchemy.org/trac/wiki/SqlSoup>`_.
-
.. automodule:: sqlalchemy.ext.sqlsoup
- :members:
- :undoc-members:
+ :members:
+
:members:
:show-inheritance:
-.. autoclass:: DefaultRunner
- :members:
- :show-inheritance:
-
.. autoclass:: ExecutionContext
:members:
After ``remove()`` is called, the next operation with the contextual session will start a new ``Session`` for the current thread.
+.. _session_lifespan:
+
Lifespan of a Contextual Session
--------------------------------
Once again, SQLAlchemy figured out the FROM clause for our statement. In fact it will determine the FROM clause based on all of its other bits; the columns clause, the where clause, and also some other elements which we haven't covered yet, which include ORDER BY, GROUP BY, and HAVING.
+.. _sqlexpression_text:
+
Using Text
===========
-
Our last example really became a handful to type. Going from what one understands to be a textual SQL expression into a Python construct which groups components together in a programmatic style can be hard. That's why SQLAlchemy lets you just use strings too. The ``text()`` construct represents any textual statement. To use bind parameters with ``text()``, always use the named colon format. Such as below, we create a ``text()`` and execute it, feeding in the bind parameters to the ``execute()`` method:
.. sourcecode:: pycon+sql
Introduction
============
-SqlSoup provides a convenient way to access database tables without
-having to declare table or mapper classes ahead of time.
+SqlSoup provides a convenient way to access existing database tables without
+having to declare table or mapper classes ahead of time. It is built on top of the SQLAlchemy ORM and provides a super-minimalistic interface to an existing database.
Suppose we have a database with users, books, and loans tables
-(corresponding to the PyWebOff dataset, if you're curious). For
-testing purposes, we'll create this db as follows::
-
- >>> from sqlalchemy import create_engine
- >>> e = create_engine('sqlite:///:memory:')
- >>> for sql in _testsql: e.execute(sql) #doctest: +ELLIPSIS
- <...
+(corresponding to the PyWebOff dataset, if you're curious).
Creating a SqlSoup gateway is just like creating an SQLAlchemy
engine::
>>> from sqlalchemy.ext.sqlsoup import SqlSoup
>>> db = SqlSoup('sqlite:///:memory:')
-or, you can re-use an existing metadata or engine::
+or, you can re-use an existing engine::
- >>> db = SqlSoup(MetaData(e))
+ >>> db = SqlSoup(engine)
You can optionally specify a schema within the database for your
SqlSoup::
- # >>> db.schema = myschemaname
-
+ >>> db.schema = myschemaname
Loading objects
===============
------------------------
Get, filter, filter_by, order_by, limit, and the rest of the
-query methods are explained in detail in the `SQLAlchemy documentation`__.
-
-__ http://www.sqlalchemy.org/docs/05/ormtutorial.html#datamapping_querying
-
+query methods are explained in detail in :ref:`ormtutorial_querying`.
Modifying objects
=================
>>> user = _
>>> user.email = 'basepair+nospam@example.edu'
- >>> db.flush()
+ >>> db.commit()
(SqlSoup leverages the sophisticated SQLAlchemy unit-of-work code, so
multiple updates to a single object will be turned into a single
-``UPDATE`` statement when you flush.)
+``UPDATE`` statement when you commit.)
To finish covering the basics, let's insert a new loan, then delete
it::
>>> book_id = db.books.filter_by(title='Regional Variation in Moss').first().id
>>> db.loans.insert(book_id=book_id, user_name=user.name)
MappedLoans(book_id=2,user_name=u'Bhargan Basepair',loan_date=None)
- >>> db.flush()
>>> loan = db.loans.filter_by(book_id=2, user_name='Bhargan Basepair').one()
>>> db.delete(loan)
- >>> db.flush()
+ >>> db.commit()
You can also delete rows that have not been loaded as objects. Let's
do our insert/delete cycle once more, this time using the loans
>>> db.loans.insert(book_id=book_id, user_name=user.name)
MappedLoans(book_id=2,user_name=u'Bhargan Basepair',loan_date=None)
- >>> db.flush()
>>> db.loans.delete(db.loans.book_id==2)
You can similarly update multiple rows at once. This will change the
>>> del db._cache['users']
>>> db.users.relate('loans', db.loans, order_by=db.loans.loan_date, cascade='all, delete-orphan')
-
Advanced Use
============
-Accessing the Session
----------------------
+Sessions, Transations and Application Integration
+-------------------------------------------------
+
+**Note:** please read and understand this section thoroughly before using SqlSoup in any web application.
SqlSoup uses a ScopedSession to provide thread-local sessions. You
can get a reference to the current one like this::
+ >>> session = db.session
+
+The default session is available at the module level in SQLSoup, via::
+
>>> from sqlalchemy.ext.sqlsoup import Session
- >>> session = Session()
+
+The configuration of this session is ``autoflush=True``, ``autocommit=False``.
+This means when you work with the SqlSoup object, you need to call ``db.commit()``
+in order to have changes persisted. You may also call ``db.rollback()`` to
+roll things back.
-Now you have access to all the standard session-based SA features,
-such as transactions. (SqlSoup's ``flush()`` is normally
-transactionalized, but you can perform manual transaction management
-if you need a transaction to span multiple flushes.)
+Since the SqlSoup object's Session automatically enters into a transaction as soon
+as it's used, it is *essential* that you call ``commit()`` or ``rollback()``
+on it when the work within a thread completes. This means all the guidelines
+for web application integration at :ref:`session_lifespan` must be followed.
+
+The SqlSoup object can have any session or scoped session configured onto it.
+This is of key importance when integrating with existing code or frameworks
+such as Pylons. If your application already has a ``Session`` configured,
+pass it to your SqlSoup object::
+
+ >>> from myapplication import Session
+ >>> db = SqlSoup(session=Session)
+
+If the ``Session`` is configured with ``autocommit=True``, use ``flush()``
+instead of ``commit()`` to persist changes - in this case, the ``Session``
+closes out its transaction immediately and no external management is needed. ``rollback()`` is also not available. Configuring a new SQLSoup object in "autocommit" mode looks like::
+
+ >>> from sqlalchemy.orm import scoped_session, sessionmaker
+ >>> db = SqlSoup('sqlite://', session=scoped_session(sessionmaker(autoflush=False, expire_on_commit=False, autocommit=True)))
Mapping arbitrary Selectables
Raw SQL
-------
-SqlSoup works fine with SQLAlchemy's `text block support`__.
-
-__ http://www.sqlalchemy.org/docs/04/sqlexpression.html#sql_text
+SqlSoup works fine with SQLAlchemy's text construct, described in :ref:`sqlexpression_text`.
+You can also execute textual SQL directly using the `execute()` method,
+which corresponds to the `execute()` method on the underlying `Session`.
+Expressions here are expressed like ``text()`` constructs, using named parameters
+with colons::
-You can also access the SqlSoup's `engine` attribute to compose SQL
-directly. The engine's ``execute`` method corresponds to the one of a
-DBAPI cursor, and returns a ``ResultProxy`` that has ``fetch`` methods
-you would also see on a cursor::
-
- >>> rp = db.bind.execute('select name, email from users order by name')
+ >>> rp = db.execute('select name, email from users where name like :name order by name', name='%Bhargan%')
>>> for name, email in rp.fetchall(): print name, email
Bhargan Basepair basepair+nospam@example.edu
- Joe Student student@example.edu
-You can also pass this engine object to other SQLAlchemy constructs.
+Or you can get at the current transaction's connection using `connection()`. This is the
+raw connection object which can accept any sort of SQL expression or raw SQL string passed to the database::
+
+ >>> conn = db.connection()
+ >>> conn.execute("'select name, email from users where name like ? order by name'", '%Bhargan%')
Dynamic table names
default schema is used.
-Extra tests
-===========
-
-Boring tests here. Nothing of real expository value.
-
-::
-
- >>> db.users.filter_by(classname=None).order_by(db.users.name).all()
- [MappedUsers(name=u'Bhargan Basepair',email=u'basepair+nospam@example.edu',password=u'basepair',classname=None,admin=1), MappedUsers(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0)]
-
- >>> db.nopk
- Traceback (most recent call last):
- ...
- PKNotFoundError: table 'nopk' does not have a primary key defined [columns: i]
-
- >>> db.nosuchtable
- Traceback (most recent call last):
- ...
- NoSuchTableError: nosuchtable
-
- >>> years_with_count.insert(published_year='2007', n=1)
- Traceback (most recent call last):
- ...
- InvalidRequestError: SQLSoup can only modify mapped Tables (found: Alias)
-
- [tests clear()]
- >>> db.loans.count()
- 1
- >>> _ = db.loans.insert(book_id=1, user_name='Bhargan Basepair')
- >>> db.expunge_all()
- >>> db.flush()
- >>> db.loans.count()
- 1
"""
from sqlalchemy import Table, MetaData, join
from sqlalchemy import schema, sql
-from sqlalchemy.orm import scoped_session, sessionmaker, mapper, class_mapper, relation
-from sqlalchemy.exceptions import SQLAlchemyError, InvalidRequestError
+from sqlalchemy.engine.base import Engine
+from sqlalchemy.orm import scoped_session, sessionmaker, mapper, class_mapper, relation, session
+from sqlalchemy.exceptions import SQLAlchemyError, InvalidRequestError, ArgumentError
from sqlalchemy.sql import expression
-_testsql = """
-CREATE TABLE books (
- id integer PRIMARY KEY, -- auto-increments in sqlite
- title text NOT NULL,
- published_year char(4) NOT NULL,
- authors text NOT NULL
-);
-
-CREATE TABLE users (
- name varchar(32) PRIMARY KEY,
- email varchar(128) NOT NULL,
- password varchar(128) NOT NULL,
- classname text,
- admin int NOT NULL -- 0 = false
-);
-
-CREATE TABLE loans (
- book_id int PRIMARY KEY REFERENCES books(id),
- user_name varchar(32) references users(name)
- ON DELETE SET NULL ON UPDATE CASCADE,
- loan_date datetime DEFAULT current_timestamp
-);
-
-insert into users(name, email, password, admin)
-values('Bhargan Basepair', 'basepair@example.edu', 'basepair', 1);
-insert into users(name, email, password, admin)
-values('Joe Student', 'student@example.edu', 'student', 0);
-
-insert into books(title, published_year, authors)
-values('Mustards I Have Known', '1989', 'Jones');
-insert into books(title, published_year, authors)
-values('Regional Variation in Moss', '1971', 'Flim and Flam');
-
-insert into loans(book_id, user_name, loan_date)
-values (
- (select min(id) from books),
- (select name from users where name like 'Joe%'),
- '2006-07-12 0:0:0')
-;
-
-CREATE TABLE nopk (
- i int
-);
-""".split(';')
__all__ = ['PKNotFoundError', 'SqlSoup']
-Session = scoped_session(sessionmaker(autoflush=True))
+Session = scoped_session(sessionmaker(autoflush=True, autocommit=False))
class PKNotFoundError(SQLAlchemyError):
pass
def insert(cls, **kwargs):
_ddl_error(cls)
- def delete(cls, *args, **kwargs):
- _ddl_error(cls)
-
- def update(cls, whereclause=None, values=None, **kwargs):
- _ddl_error(cls)
-
def __clause_element__(cls):
return cls._table
o.__dict__.update(kwargs)
return o
- def delete(cls, *args, **kwargs):
- cls._table.delete(*args, **kwargs).execute()
-
- def update(cls, whereclause=None, values=None, **kwargs):
- cls._table.update(whereclause, values).execute(**kwargs)
-
def relate(cls, propname, *args, **kwargs):
class_mapper(cls)._configure_property(propname, relation(*args, **kwargs))
mappr = mapper(klass,
selectable,
extension=Session.extension,
- allow_null_pks=_is_outer_join(selectable),
**mapper_kwargs)
for k in mappr.iterate_properties:
klass._query = Session.query_property()
return klass
-class SqlSoup:
- def __init__(self, *args, **kwargs):
+class SqlSoup(object):
+ def __init__(self, engine_or_metadata, **kw):
"""Initialize a new ``SqlSoup``.
`args` may either be an ``SQLEngine`` or a set of arguments
suitable for passing to ``create_engine``.
"""
-
- # meh, sometimes having method overloading instead of kwargs would be easier
- if isinstance(args[0], MetaData):
- args = list(args)
- metadata = args.pop(0)
- if args or kwargs:
- raise ArgumentError('Extra arguments not allowed when metadata is given')
+
+ self.session = kw.pop('session', Session)
+
+ if isinstance(engine_or_metadata, MetaData):
+ self._metadata = engine_or_metadata
+ elif isinstance(engine_or_metadata, (basestring, Engine)):
+ self._metadata = MetaData(engine_or_metadata)
else:
- metadata = MetaData(*args, **kwargs)
- self._metadata = metadata
+ raise ArgumentError("invalid engine or metadata argument %r" % engine_or_metadata)
+
self._cache = {}
self.schema = None
-
+
def engine(self):
return self._metadata.bind
bind = engine
def delete(self, *args, **kwargs):
- Session.delete(*args, **kwargs)
-
+ self.session.delete(*args, **kwargs)
+
+ def execute(self, stmt, **params):
+ return self.session.execute(sql.text(stmt, bind=self.bind), **params)
+
+ @property
+ def _underlying_session(self):
+ if isinstance(self.session, session.Session):
+ return self.session
+ else:
+ return self.session()
+
+ def connection(self):
+ return self._underlying_session._connection_for_bind(self.bind)
+
def flush(self):
- Session.flush()
-
+ self.session.flush()
+
+ def rollback(self):
+ self.session.rollback()
+
+ def commit(self):
+ self.session.commit()
+
def clear(self):
- Session.expunge_all()
-
+ self.session.expunge_all()
+
+ def expunge(self, *args, **kw):
+ self.session.expunge(*args, **kw)
+
def expunge_all(self):
- Session.expunge_all()
+ self.session.expunge_all()
def map(self, selectable, **kwargs):
try:
def with_labels(self, item):
# TODO give meaningful aliases
- return self.map(expression._clause_element_as_expr(item).select(use_labels=True).alias('foo'))
+ return self.map(
+ expression._clause_element_as_expr(item).
+ select(use_labels=True).
+ alias('foo'))
def join(self, *args, **kwargs):
j = join(*args, **kwargs)
try:
t = self._cache[attr]
except KeyError, ke:
- table = Table(attr, self._metadata, autoload=True, schema=schema or self.schema)
+ table = Table(attr, self._metadata, autoload=True, autoload_with=self.bind, schema=schema or self.schema)
if not table.primary_key.columns:
- # Py3K
- #raise PKNotFoundError('table %r does not have a primary key defined [columns: %s]' % (attr, ','.join(table.c.keys()))) from ke
- # Py2K
raise PKNotFoundError('table %r does not have a primary key defined [columns: %s]' % (attr, ','.join(table.c.keys())))
- # end Py2K
if table.columns:
t = class_for_table(table)
else:
def __repr__(self):
return 'SqlSoup(%r)' % self._metadata
-if __name__ == '__main__':
- import logging
- logging.basicConfig()
- import doctest
- doctest.testmod()
Optional, any ``ClauseElement``
"""
- return self.__connection(self.get_bind(mapper, clause))
+ return self._connection_for_bind(self.get_bind(mapper, clause))
- def __connection(self, engine, **kwargs):
+ def _connection_for_bind(self, engine, **kwargs):
if self.transaction is not None:
return self.transaction._connection_for_bind(engine)
else:
engine = self.get_bind(mapper, clause=clause, **kw)
- return self.__connection(engine, close_with_result=True).execute(
+ return self._connection_for_bind(engine, close_with_result=True).execute(
clause, params or {})
def scalar(self, clause, params=None, mapper=None, **kw):
--- /dev/null
+from sqlalchemy.ext import sqlsoup
+from sqlalchemy.test.testing import TestBase, eq_, assert_raises
+from sqlalchemy import create_engine, or_, desc, select, func, exc
+import datetime
+
+class SQLSoupTest(TestBase):
+ @classmethod
+ def setup_class(cls):
+ global engine
+ engine = create_engine('sqlite://')
+ for sql in _ddl:
+ engine.execute(sql)
+
+ @classmethod
+ def teardown_class(cls):
+ engine.dispose()
+
+ def setup(self):
+ for sql in _data:
+ engine.execute(sql)
+
+ def teardown(self):
+ sqlsoup.Session.remove()
+ for sql in _teardown:
+ engine.execute(sql)
+
+
+ def test_load(self):
+ db = sqlsoup.SqlSoup(engine)
+ MappedUsers = db.users
+ users = db.users.all()
+ users.sort()
+ eq_(
+ users,
+ [
+ MappedUsers(
+ name=u'Joe Student',
+ email=u'student@example.edu',
+ password=u'student',classname=None,admin=0),
+ MappedUsers(
+ name=u'Bhargan Basepair',
+ email=u'basepair@example.edu',
+ password=u'basepair',classname=None,admin=1)
+ ]
+ )
+
+ def test_order_by(self):
+ db = sqlsoup.SqlSoup(engine)
+ MappedUsers = db.users
+ users = db.users.order_by(db.users.name).all()
+ eq_(
+ users,
+ [
+ MappedUsers(
+ name=u'Bhargan Basepair',
+ email=u'basepair@example.edu',
+ password=u'basepair',classname=None,admin=1),
+ MappedUsers(
+ name=u'Joe Student',
+ email=u'student@example.edu',
+ password=u'student',classname=None,admin=0),
+ ]
+ )
+
+ def test_whereclause(self):
+ db = sqlsoup.SqlSoup(engine)
+ MappedUsers = db.users
+
+ where = or_(db.users.name=='Bhargan Basepair', db.users.email=='student@example.edu')
+ users = db.users.filter(where).order_by(desc(db.users.name)).all()
+ eq_(
+ users,
+ [MappedUsers(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0), MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1)]
+ )
+
+ def test_first(self):
+ db = sqlsoup.SqlSoup(engine)
+ MappedUsers = db.users
+
+ user = db.users.filter(db.users.name=='Bhargan Basepair').one()
+ eq_(user,
+ MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1)
+ )
+ db.rollback()
+
+ user = db.users.get('Bhargan Basepair')
+ eq_(user, MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1))
+ db.rollback()
+
+ user = db.users.filter_by(name='Bhargan Basepair').one()
+ eq_(user, MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1))
+ db.rollback()
+
+ def test_crud(self):
+ # note we're testing autoflush here too...
+ db = sqlsoup.SqlSoup(engine)
+ MappedLoans = db.loans
+ user = db.users.filter_by(name='Bhargan Basepair').one()
+ book_id = db.books.filter_by(title='Regional Variation in Moss').first().id
+ loan_insert = db.loans.insert(book_id=book_id, user_name=user.name)
+
+ loan = db.loans.filter_by(book_id=2, user_name='Bhargan Basepair').one()
+ eq_(loan, loan_insert)
+ l2 = MappedLoans(book_id=2,user_name=u'Bhargan Basepair',loan_date=loan.loan_date)
+ eq_(loan, l2)
+ db.expunge(l2)
+
+ db.delete(loan)
+ loan = db.loans.filter_by(book_id=2, user_name='Bhargan Basepair').first()
+ assert loan is None
+
+ def test_cls_crud(self):
+ db = sqlsoup.SqlSoup(engine)
+ MappedUsers = db.users
+ db.users.filter_by(name='Bhargan Basepair').update(dict(name='Some New Name'))
+ u1= db.users.filter_by(name='Some New Name').one()
+ eq_(
+ u1,
+ MappedUsers(name=u'Some New Name',
+ email=u'basepair@example.edu',
+ password=u'basepair',classname=None,admin=1)
+ )
+
+
+ def test_mapped_join(self):
+ db = sqlsoup.SqlSoup(engine)
+
+ join1 = MappedJoin = db.join(db.users, db.loans, isouter=True)
+ mj = join1.filter_by(name='Joe Student').all()
+ eq_(
+ mj,
+ [MappedJoin(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0,book_id=1,user_name=u'Joe Student',loan_date=datetime.datetime(2006, 7, 12, 0, 0))]
+ )
+
+ db.rollback()
+
+ join2 = MappedJoin = db.join(join1, db.books)
+ mj = join2.all()
+ eq_(mj, [MappedJoin(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0,book_id=1,user_name=u'Joe Student',loan_date=datetime.datetime(2006, 7, 12, 0, 0),id=1,title=u'Mustards I Have Known',published_year=u'1989',authors=u'Jones')])
+
+ eq_(
+ db.with_labels(join1).c.keys(),
+ [u'users_name', u'users_email', u'users_password',
+ u'users_classname', u'users_admin',
+ u'loans_book_id', u'loans_user_name',
+ u'loans_loan_date']
+ )
+
+ labeled_loans = db.with_labels(db.loans)
+ eq_(
+ db.join(db.users, labeled_loans, isouter=True).c.keys(),
+ [u'name', u'email', u'password', u'classname',
+ u'admin', u'loans_book_id', u'loans_user_name', u'loans_loan_date']
+ )
+
+ def test_relations(self):
+ db = sqlsoup.SqlSoup(engine)
+ db.users.relate('loans', db.loans)
+
+ MappedLoans = db.loans
+ MappedUsers = db.users
+
+ eq_(
+ db.users.get('Joe Student').loans,
+ [MappedLoans(
+ book_id=1,
+ user_name=u'Joe Student',
+ loan_date=datetime.datetime(2006, 7, 12, 0, 0))
+ ]
+ )
+ db.rollback()
+
+ eq_(
+ db.users.filter(~db.users.loans.any()).all(),
+ [MappedUsers(
+ name=u'Bhargan Basepair',
+ email='basepair@example.edu',
+ password=u'basepair',
+ classname=None,admin=1)
+ ]
+ )
+ db.rollback()
+
+ del db._cache['users']
+ db.users.relate('loans', db.loans,
+ order_by=db.loans.loan_date, cascade='all, delete-orphan')
+
+
+ def test_selectable(self):
+ db = sqlsoup.SqlSoup(engine)
+ MappedBooks = db.books
+ b = db.books._table
+
+ s = select(
+ [b.c.published_year, func.count('*').label('n')],
+ from_obj=[b], group_by=[b.c.published_year])
+
+ s = s.alias('years_with_count')
+ years_with_count = db.map(s, primary_key=[s.c.published_year])
+
+ eq_(
+ years_with_count.filter_by(published_year='1989').all(),
+ [MappedBooks(published_year=u'1989',n=1)]
+ )
+
+ def test_raw_sql(self):
+ db = sqlsoup.SqlSoup(engine)
+ rp = db.execute('select name, email from users order by name')
+ eq_(
+ rp.fetchall(),
+ [('Bhargan Basepair', 'basepair@example.edu'),
+ ('Joe Student', 'student@example.edu')]
+ )
+
+ # test that execute() shares the same transactional
+ # context as the session
+ db.execute("update users set email='foo bar'")
+ eq_(
+ db.execute("select distinct email from users").fetchall(),
+ [('foo bar',)]
+ )
+ db.rollback()
+ eq_(
+ db.execute("select distinct email from users").fetchall(),
+ [(u'basepair@example.edu',), (u'student@example.edu',)]
+ )
+
+ def test_connection(self):
+ db = sqlsoup.SqlSoup(engine)
+ conn = db.connection()
+ rp = conn.execute('select name, email from users order by name')
+ eq_(
+ rp.fetchall(),
+ [('Bhargan Basepair', 'basepair@example.edu'),
+ ('Joe Student', 'student@example.edu')]
+ )
+
+ def test_entity(self):
+ db = sqlsoup.SqlSoup(engine)
+ tablename = 'loans'
+ eq_(db.entity(tablename), db.loans)
+
+ def test_filter_by_order_by(self):
+ db = sqlsoup.SqlSoup(engine)
+ MappedUsers = db.users
+ users = db.users.filter_by(classname=None).order_by(db.users.name).all()
+ eq_(
+ users,
+ [MappedUsers(name=u'Bhargan Basepair',email=u'basepair@example.edu',password=u'basepair',classname=None,admin=1), MappedUsers(name=u'Joe Student',email=u'student@example.edu',password=u'student',classname=None,admin=0)]
+ )
+
+ def test_no_pk(self):
+ db = sqlsoup.SqlSoup(engine)
+ assert_raises(sqlsoup.PKNotFoundError, getattr, db, 'nopk')
+
+ def test_nosuchtable(self):
+ db = sqlsoup.SqlSoup(engine)
+ assert_raises(exc.NoSuchTableError, getattr, db, 'nosuchtable')
+
+ def test_dont_persist_alias(self):
+ db = sqlsoup.SqlSoup(engine)
+ MappedBooks = db.books
+ b = db.books._table
+
+ s = select(
+ [b.c.published_year, func.count('*').label('n')],
+ from_obj=[b], group_by=[b.c.published_year])
+
+ s = s.alias('years_with_count')
+ years_with_count = db.map(s, primary_key=[s.c.published_year])
+
+ assert_raises(exc.InvalidRequestError, years_with_count.insert, published_year='2007', n=1)
+
+ def test_clear(self):
+ db = sqlsoup.SqlSoup(engine)
+ eq_(db.loans.count(), 1)
+ _ = db.loans.insert(book_id=1, user_name='Bhargan Basepair')
+ db.expunge_all()
+ db.flush()
+ eq_(db.loans.count(), 1)
+
+_ddl = """
+CREATE TABLE books (
+ id integer PRIMARY KEY, -- auto-increments in sqlite
+ title text NOT NULL,
+ published_year char(4) NOT NULL,
+ authors text NOT NULL
+);
+
+CREATE TABLE users (
+ name varchar(32) PRIMARY KEY,
+ email varchar(128) NOT NULL,
+ password varchar(128) NOT NULL,
+ classname text,
+ admin int NOT NULL -- 0 = false
+);
+
+CREATE TABLE loans (
+ book_id int PRIMARY KEY REFERENCES books(id),
+ user_name varchar(32) references users(name)
+ ON DELETE SET NULL ON UPDATE CASCADE,
+ loan_date datetime DEFAULT current_timestamp
+);
+
+
+CREATE TABLE nopk (
+ i int
+);
+""".split(';')
+
+
+_data = """
+insert into users(name, email, password, admin)
+values('Bhargan Basepair', 'basepair@example.edu', 'basepair', 1);
+insert into users(name, email, password, admin)
+values('Joe Student', 'student@example.edu', 'student', 0);
+
+insert into books(title, published_year, authors)
+values('Mustards I Have Known', '1989', 'Jones');
+insert into books(title, published_year, authors)
+values('Regional Variation in Moss', '1971', 'Flim and Flam');
+
+insert into loans(book_id, user_name, loan_date)
+values (
+ (select min(id) from books),
+ (select name from users where name like 'Joe%'),
+ '2006-07-12 0:0:0')
+;
+""".split(";")
+
+_teardown = """
+delete from loans;
+delete from books;
+delete from users;
+delete from nopk;
+""".split(";")
\ No newline at end of file
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
