Helpers SQLAlchemy - `sqla_helpers.base_model.BaseModel`¶

Installation¶

Git

Installation from git

$> git clone git@github.com:moumoutte/sqla_helpers.git
$> cd sqla_helpers
$> sudo python2.7 setup.py install

Eggs

Installation from pypi eggs

$> sudo pip install sqla_helpers

Getting Started¶

The goal of sqla_helpers.base_model.BaseModel is to provide syntactic sugar for SQLAlchemy.

sqla_helpers.base_model.BaseModel is to use as mixin class. This class inherits from nothing and shouldn’t be inherited. For access method from model, models need to be declared as bellow:

from somewhere import DeclarativeBase
from sqla_helpers.base_model import BaseModel

class MyModel(DeclarativeBase, BaseModel):
    id = ... # Clef primaire , l'identifiant sous forme d'entier
    awesome_attr = ... # Attribut quelconque du modèle
    other_model = relationship('MyOtherModel', backref='mymodel')


class MyOtherModel(DeclarativeBase, BaseModel):
    id = ... # Clef primaire
    name = ...
    model_id = ... # Clef étrangère sur MyModel

The DeclarativeBase class is generated by declarative_base() function from SQLAlchemy.

To avoid mixin uses, sqla_helpers.base_model.BaseModel class can be used as cls parameter in declarative_base() function.

from sqlalchemy.ext.declarative import declarative_base
from sqla_helpers.base_model import BaseModel
DeclarativeBase = declarative_base(cls=BaseModel)

class MyModel(DeclarativeBase):
    # ...

sqla_helpers.base_model.BaseModel needs to build a session when queries are done. In order to access a session when needing, the class uses the stored function sqla_helpers.base_model.BaseModel.sessionmaker. This function will be called each time a session is needed. So we need to store a session_maker by calling sqla_helpers.base_model.BaseModel.register_sessionmaker method.

# Application's initialization
def main():
    # ...
    BaseModel.register_sessionmaker(scoped_session(sessionmaker(bind=engine)))
    # ...

For a global session, you can just give a Session which is not a callable

from somwhere import DBSession

# Application's initialization
def main():
    # ...
    BaseModel.register_sessionmaker(DBSession)
    # ...

Registering a session maker can be dangerous. Because, technically, we change dynamically a class method. To prevent errors, an exception , sqla_helpers.base_model.SessionMakerExists, is raised if a session maker is already registered.

But sometimes, perhaps you need to change it while application is running. So, you can force a new record even if a session maker is already registered

>>> BaseModel.register_sessionmaker(db_session)
>>> new_db_session = amazing_function()
>>> BaseModel.register_sessionmaker(new_db_session)
Traceback (most recent call last):
    File "<stdin>", line 1, in <module>
SessionMakerExists: A session maker is already registered.
>>> BaseModel.register_sessionmaker(new_db_session, force=True)
Traceback (most recent call last):
    File "<stdin>", line 1, in <module>
SessionMakerExists: A session maker is already registered.

Basic use case :

>>> MyModel.all()
[<MyModel object at 0x2c19d90>]
>>> MyModel.get(id=2)
<MyModel object at 0x2c19d90>
>>> MyModel.get(id=3)
*** NoResultFound: No row was found for one()
>>> MyModel.filter(id=2)
[<MyModel object at 0x2c19d90>]
>>> MyModel.filter(id=3)
[]
>>> MyModel.count(id=2)
1

sqla_helpers.base_model.BaseModel.all() returns all the database objects
sqla_helpers.base_model.BaseModel.filter() returns a list of matching objects.
sqla_helpers.base_model.BaseModel.get() returns an uniq matching object.
sqla_helpers.base_model.BaseModel.count() returns the number of matching objects.

Querying criterions can be chained with an && (logical and) operator.

>>> MyOtherModel.filter(name='toto')
[<MyOtherModel object at 0x2c19d90>, <MyOtherModel object at 0x2e27e08>]
>>> MyOtherModel.filter(name='toto', id=2)
[<MyOtherModel object at 0x2c19d90>]

Querying for criterions on relations¶

Valid querying criterions for a class are defined by the class attributes. IE : in case of MyOtherModel, criterions can be id, name and model_id.

This is still true for a Sqlachemy relation.

IE: querying all MyModel witch MyOtherModel have a name ‘foo’.

>>> MyModel.filter(awesome_attr__name='foo')
[<MyModel object at 0x2c19d90>]

Querying with entire object.

>>> otherModel = MyOtherModel.get(name='foo')
>>> MyModel.filter(awesome_attr=otherModel)
[<MyModel object at 0x2c19d90>]

The __ separator (double underscore) allows to split between the differents entities.

Quering with relations attributes can be done recursively. If MyOtherObject has an other_attr attribute which is in relation with a MyOtherOtherObject object.

Querying all MyModel with a MyOtherObject has MyOtherOtherObject has a name attribute is ‘foo’.

>>> MyModel.filter(awesome_attr__other_attr__name='foo')
[<MyModel object at 0x2c19d90>]

Operators¶

Others operators than equality can be used. Those operators should be written with the attribute name following ‘__’ (double underscore) and operator’s name.

IE: if all MyModel with id different than 2 are wanted:

>>> MyModel.filter(id__not=2)
[]

Available operators are:

‘not’: Non-equal,
‘lt’: letter than,
‘le’: letter or equals than,
‘gt’: gretter than,
‘ge’: gretter or equal than,
‘in’: in a list,
‘like’: SQL LIKE operator,
‘ilike’: SQL ILIKE operator.

More complex querying¶

As the Django way, sqla_helpers provides a sqla_helpers.logical.Q object for more complex queries. The sqla_helpers.logical.Q object can use the :mod:`sqla_helpers’ syntax.

>>> from sqla_helpers.logical import Q
>>> Q(status__name='test')
<sqla_helpers.logical.Q at 0x2376cd0>

These objects are usable as criterions for query.

sqla_helpers.base_model.BaseModel

>>> Treatment.get(Q(id=2))
>>> <sqlalchemy_test.models.Treatment at 0x2388690>

The goal of those objects is to allow SQL logical conditions in a python syntax.

If all Treatment objects wih an id == 2 or a Status name == ‘KO’ are wanted.

>>>  Treatment.filter(Q(id=2) | Q(status__name='KO'))
[<sqlalchemy_test.models.Treatment at 0x2388690>, <sqlalchemy_test.models.Treatment at 0x23837d0>]

For getting, all Treatment objects with an `id’ attribute different than 2 :

>>> Treatment.filter(~Q(id=2))
[<sqlalchemy_test.models.Treatment at 0x2383450>, <sqlalchemy_test.models.Treatment at 0x23837d0>,
  <sqlalchemy_test.models.Treatment at 0x23886d0> ]

Logical operators can be chained :

>>> Treatment.filter((Q(id=2) | Q(name='toto')) & (Q(name='OK') | ~Q(status__id=3)))
2013-02-10 16:39:49,485 INFO sqlalchemy.engine.base.Engine SELECT
treatment.id AS treatment_id, treatment.name AS treatment_name,
treatment.status_id AS treatment_status_id
FROM treatment JOIN status ON status.id = treatment.status_id
WHERE (treatment.id = ? OR treatment.name = ?) AND (treatment.name = ? OR
status.id != ?)
2013-02-10 16:39:49,485 INFO sqlalchemy.engine.base.Engine (2, 'toto', 'OK',
3)
>>> [<sqlalchemy_test.models.Treatment at 0x2388690>]

JSON¶

Often in web oriented applications, client and server exchange with JSON format. In order to have easier loading, sqla_helpers provides methods for loading from a regular python dictionary or a SQLAlchemy model object.

The sqla_helpers.base_model.BaseModel.dump() method allows a JSON compatible dictionary.

>>> print json.dumps(t.dump(), indent=4)
{
    "status": {
        "id": 1,
        "name": "Ok"
    },
    "status_id": 1,
    "id": 1,
    "name": "Great Treatment"
}

The method sqla_helpers.base_model.BaseModel.load can build objects from a dictionary. The meaning of use a dictionary is to facilitate access to data in JSON or generate JSON from dictionary.

Objects are getting from database if primary key attributes are found on the dictionnary. Otherwise new object are created.

>>> t = Treatment.get(id=7)
>>> t.name
'YEAH \\o/'
>>> t.id
7
>>> t.status.name
'Holy status !'
>>> t.status.id
7
>>> t = Treatment.load({'id': 7, 'name': 'hello'})
>>> t.name, t.id
('hello', 7)
>>> session.commit()
>>> t.dump()
{
        'id': 7,
        'name': u'hello',
        'status': {'id': 7, 'name': u'Holy status !'},
        'status_id': 7
}
>>> tr = Treatment.load(t.dump())
>>> tr == t
True
>>> tr.status == t.status
True
>>> Treatment.load(tr.dump()).dump()
{
        'id': 7,
        'name': u'hello',
        'status': {'id': 7, 'name': u'Holy status !'},
        'status_id': 7
}
>>> tr = Treatment.load({'name': 'new treatment', 'status': {'name': 'new status'}})
>>> tr.id
None
>>> tr.status.id
None
>>> session.add(tr)
>>> session.commit()
>>> tr.id
10
>>> tr.status.id
8

`sqla_helpers.base_model.BaseModel` class¶

class sqla_helpers.base_model.SessionMakerExists¶: Exception raised when a session maker is already registered in sqla_helpers.base_model.BaseModel

sqla_helpers.base_model.query_operation(decorated_method=None, operation_name=None)¶

query_operation process a search on a query with criterion and operators. Then, the operation_name operation is called on the query. If operation_name is not set the operation_name is taken from the decorated method

@query_operation
def count(cls, *operators, **criterions):
    # This method will call the cls.search(*operators, **criterions)
    # Then , it will call count operation on the result of previous search
    # I.E. : the generated code will be "return cls.search(*operators, **criterions).count()"

@query_operation(operation_name='one')
def get(cls, *operators, **criterions):
    # It will process the same thing above, except the function compute on the query will be 'one'
    # I.E. : return cls.search(*operators, **criterions).one()

class sqla_helpers.base_model.BaseModel¶

Base Model Class. Provide syntactic sugar for getting object from database.

classmethod all(querying_class, *operators, **criterions)¶: Returns all objects from the same class contained in database.

classmethod count(querying_class, *operators, **criterions)¶

Returns the number of objects matched by criterions

>>> Treatment.count(status=u'OK')
8

dump(excludes=, []depth=2)¶

Returns object as dictionary with dependencies.

Depth limits the recursion.

IE : With depth set as 1, objects in relations aren’t search.

excludes use to exclude unwanted attributes.

>>> t = Treatment.get(id=1)
>>> print json.dumps(t.dump(), indent=4)
{
    "status": {
        "id": 1,
        "name": "Ok"
    },
    "status_id": 1,
    "id": 1,
    "name": "Great Treatment"
}

>>> print json.dumps(t.dump(depth=1), indent=4)
{
    "status_id": 1,
    "id": 1,
    "name": "Great Treatment"
}
>>> print json.dumps(t.dump(excludes=['status_id']), indent=4)
{
    "status": {
        "id": 1,
        "name": "Ok"
    },
    "id": 1,
    "name": "Great Treatment"
}

classmethod filter(querying_class, *operators, **criterions)¶: Returns a list of objects from a class matching criterions given in parameters.

classmethod get(querying_class, *operators, **criterions)¶: Returns an object with criterions given in parameters.

classmethod load(dictionary, hard=False)¶

Returns an object from class with attributes got in dictionary’s parameters.

If all the primary keys are found in the dictionary, the object is loaded from database. Otherwise, values are set in the loading object.

>>> t = Treatment.get(id=1)
>>> t.name
'Great Treatment'
>>> t = Treatment.load({'id': 1, 'name': 'Awesome Treatment'})
>>> t.name
'Awesome Treatment'
>>> session.commit()
>>> Treatment.get(id=1).name
'Awesome Treatment'

If hard parameter is True, an exception is raised if a value isn’t found in parameter’s dictionary.

classmethod process_params(class_found, **kwargs)¶

Returns a SQLAlchemy criterions list matching sqla_helpers syntax.

:param:`cls` is the root class providing attributes

During processing, found attributes are stored in :param:`class_found` parameters. :param:`class_found` is a set.

Attribute is updated during process.

Example

If a quering on the attribute name from a Treatment object, the function ‘ll be called:

>>> class_found = []
>>> process_params(Treatment, class_found, name='test')
[<sqlalchemy.sql.expression.BinaryExpression object at 0x22bd3d0>]
>>> class_found
[]

In this example, because of the attribute name isn’t in the related object, the :param:`class_found` isn’t modified. In other hand a query on status attribute :

>>> class_found = []
>>> process_params(Treatment, class_found, status_name='test')
[<sqlalchemy.sql.expression.BinaryExpression object at 0x22bd3d0>]
>>> class_found
[Status]

classmethod register_sessionmaker(sessionmaker, force=False)¶: Register the function for making session. This registered function mustn’t have any parameters. For a globale session, just put the session as parameter. If a session maker is already registered, an exception is raised to avoid conflict. But, if you are sure about what your are doing, you can set force parameter to True. It’s not advice.

classmethod search(*operator, **criterion)¶

Object search with criterions given in arguments. Returns a sqlachemy.orm.query.Query object.

Filters can be chained.

Helpers SQLAlchemy - sqla_helpers.base_model.BaseModel¶