diff options
| -rw-r--r-- | CHANGES | 17 | ||||
| -rw-r--r-- | doc/build/ormtutorial.rst | 2 | ||||
| -rw-r--r-- | doc/build/reference/ext/sqlsoup.rst | 19 | ||||
| -rw-r--r-- | doc/build/reference/sqlalchemy/connections.rst | 4 | ||||
| -rw-r--r-- | doc/build/session.rst | 2 | ||||
| -rw-r--r-- | doc/build/sqlexpression.rst | 3 | ||||
| -rw-r--r-- | lib/sqlalchemy/ext/sqlsoup.py | 267 | ||||
| -rw-r--r-- | lib/sqlalchemy/orm/session.py | 6 | ||||
| -rw-r--r-- | test/ext/test_sqlsoup.py | 336 |
9 files changed, 466 insertions, 190 deletions
@@ -646,7 +646,22 @@ CHANGES - 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 diff --git a/doc/build/ormtutorial.rst b/doc/build/ormtutorial.rst index dd2711d20..855f95140 100644 --- a/doc/build/ormtutorial.rst +++ b/doc/build/ormtutorial.rst @@ -316,6 +316,8 @@ issuing a SELECT illustrates the changes made to the database: ['ed', 'fakeuser'] {stop}[<User('ed','Ed Jones', 'f8s7ccs')>] +.. _ormtutorial_querying: + Querying ======== diff --git a/doc/build/reference/ext/sqlsoup.rst b/doc/build/reference/ext/sqlsoup.rst index cd79e4cbc..fcc937166 100644 --- a/doc/build/reference/ext/sqlsoup.rst +++ b/doc/build/reference/ext/sqlsoup.rst @@ -1,21 +1,6 @@ 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: + diff --git a/doc/build/reference/sqlalchemy/connections.rst b/doc/build/reference/sqlalchemy/connections.rst index 394fa864c..f1bb1a512 100644 --- a/doc/build/reference/sqlalchemy/connections.rst +++ b/doc/build/reference/sqlalchemy/connections.rst @@ -58,10 +58,6 @@ Internals :members: :show-inheritance: -.. autoclass:: DefaultRunner - :members: - :show-inheritance: - .. autoclass:: ExecutionContext :members: diff --git a/doc/build/session.rst b/doc/build/session.rst index d8f01d5fc..640e3a330 100644 --- a/doc/build/session.rst +++ b/doc/build/session.rst @@ -606,6 +606,8 @@ The contextual session may be disposed of by calling ``Session.remove()``:: 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 -------------------------------- diff --git a/doc/build/sqlexpression.rst b/doc/build/sqlexpression.rst index 71a40e77f..7ff29c748 100644 --- a/doc/build/sqlexpression.rst +++ b/doc/build/sqlexpression.rst @@ -477,10 +477,11 @@ So with all of this vocabulary, let's select all users who have an email address 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 diff --git a/lib/sqlalchemy/ext/sqlsoup.py b/lib/sqlalchemy/ext/sqlsoup.py index 6eef4657c..592878acd 100644 --- a/lib/sqlalchemy/ext/sqlsoup.py +++ b/lib/sqlalchemy/ext/sqlsoup.py @@ -2,17 +2,11 @@ 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:: @@ -20,15 +14,14 @@ 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 =============== @@ -84,10 +77,7 @@ Full query documentation ------------------------ 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 ================= @@ -96,11 +86,11 @@ Modifying objects is intuitive:: >>> 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:: @@ -108,11 +98,10 @@ 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 @@ -125,7 +114,6 @@ to the select methods. >>> 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 @@ -203,23 +191,47 @@ relate can take any options that the relation function accepts in normal mapper >>> 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 @@ -260,21 +272,21 @@ Python is flexible like that! 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 @@ -290,95 +302,19 @@ entity() also takes an optional schema argument. If none is specified, the 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 @@ -394,12 +330,6 @@ class SelectableClassType(type): 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 @@ -415,12 +345,6 @@ class TableClassType(SelectableClassType): 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)) @@ -479,7 +403,6 @@ def class_for_table(selectable, **mapper_kwargs): mappr = mapper(klass, selectable, extension=Session.extension, - allow_null_pks=_is_outer_join(selectable), **mapper_kwargs) for k in mappr.iterate_properties: @@ -488,26 +411,26 @@ def class_for_table(selectable, **mapper_kwargs): 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 @@ -515,16 +438,38 @@ class SqlSoup: 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: @@ -536,7 +481,10 @@ class SqlSoup: 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) @@ -546,13 +494,9 @@ class SqlSoup: 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: @@ -566,8 +510,3 @@ class SqlSoup: def __repr__(self): return 'SqlSoup(%r)' % self._metadata -if __name__ == '__main__': - import logging - logging.basicConfig() - import doctest - doctest.testmod() diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py index d68840c51..c0b5666b0 100644 --- a/lib/sqlalchemy/orm/session.py +++ b/lib/sqlalchemy/orm/session.py @@ -696,9 +696,9 @@ class Session(object): 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: @@ -735,7 +735,7 @@ class Session(object): 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): diff --git a/test/ext/test_sqlsoup.py b/test/ext/test_sqlsoup.py new file mode 100644 index 000000000..8f1a45f11 --- /dev/null +++ b/test/ext/test_sqlsoup.py @@ -0,0 +1,336 @@ +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 |