API Reference

Model

Declarative base for ORM models.

class alchy.model.ModelMeta(name, bases, dct)[source]

ModelBase’s metaclass which provides:

class alchy.model.ModelBase(*args, **kargs)[source]

Base class for creating a declarative base for models.

To create a declarative base:

# in project/core.py
from alchy import ModelBase, make_declarative_base

class Base(ModelBase):
    # augument the ModelBase with super powers
    pass

Model = make_declarative_base(Base=Base)


# in project/models/user.py
from project.core import Model

class User(Model):
    # define declarative User model
    pass
__table_args__

Default table args.

__mapper_args__

Define a default order by when not specified by query operation, e.g.: { 'order_by': [column1, column2] }

__bind_key__

Bind a model to a particular database URI using keys from Manager.config['SQLALCHEMY_BINDS']. By default a model will be bound to Manager.config['SQLALCHEMY_DATABASE_URI'].

__events__

Register orm event listeners. See alchy.events for more details.

query_class

Query class to use for self.query.

query

An instance of query_class. Can be used to query the database for instances of this model. NOTE: Requires setting MyClass.query = QueryProperty(session) when session available. See make_declarative_base() for automatic implementation.

__getitem__(item)[source]

Proxy getitem to getattr.

__init__(*args, **kargs)[source]

Initialize model instance by calling update().

__iter__()[source]

Implement __iter__ so model can be converted to dict via dict().

__setitem__(item, value)[source]

Proxy setitem to setattr.

__to_dict__

Configuration for to_dict(). Do any necessary preprocessing and return a set of string attributes which represent the fields which should be returned when calling to_dict().

By default this model is refreshed if it’s __dict__ state is empty and only the ORM descriptor fields are returned.

This is the property to override if you want to return more/less than the default ORM descriptor fields.

Generally, we can usually rely on __dict__ as a representation of model when it’s just been loaded from the database. In this case, whatever values are present in __dict__ are the loaded values from the database which include/exclude lazy attributes (columns and relationships).

One issue to be aware of is that after a model has been committed (or expired), __dict__ will be empty. This can be worked around by calling refresh() which will reload the data from the database using the default loader strategies.

These are the two main cases this default implementation will try to cover. For anything more complex it would be best to override this property or the to_dict() method itself.

classmethod attrs()[source]

Return ORM attributes

classmethod column_attrs()[source]

Return table columns as list of class attributes at the class level.

classmethod columns()[source]

Return table columns.

delete(*args, **kargs)[source]

Call session.delete() on self

descriptor_dict

Return __dict__ key-filtered by descriptors.

classmethod descriptors()[source]

Return all ORM descriptors

expire(*args, **kargs)[source]

Call session.expire() on self

expunge(*args, **kargs)[source]

Call session.expunge() on self

classmethod filter(*args, **kargs)[source]

Proxy to cls.query.filter()

classmethod filter_by(*args, **kargs)[source]

Proxy to cls.query.filter_by()

flush(*args, **kargs)[source]

Call session.flush() on self

classmethod get(*args, **kargs)[source]

Proxy to cls.query.get()

classmethod get_by(data_dict=None, **kargs)[source]

Return first instance filtered by values using cls.query.filter_by().

is_modified(*args, **kargs)[source]

Call session.is_modified() on self

object_session

Return session belonging to self

classmethod primary_attrs()[source]

Return class attributes from primary keys.

classmethod primary_key()[source]

Return primary key as either single column (one primary key) or tuple otherwise.

classmethod primary_keys()[source]

Return primary keys as tuple.

query_class

alias of QueryModel

refresh(*args, **kargs)[source]

Call session.refresh() on self

classmethod relationships()[source]

Return ORM relationships

save(*args, **kargs)[source]

Call session.add() on self

classmethod session()[source]

Return session from query property

to_dict()[source]

Return dict representation of model by filtering fields using __to_dict__.

update(data_dict=None, **kargs)[source]

Update model with arbitrary set of data.

alchy.model.make_declarative_base(session=None, Model=None, Base=None)[source]

Factory function for either creating a new declarative base class or extending a previously defined one.

alchy.model.extend_declarative_base(Model, session=None)[source]

Extend a declarative base class with additional properties.

  • Extend Model with query property accessor

Query

Query subclass used by Manager as default session query class.

class alchy.query.Query(entities, session=None)[source]

Extension of default Query class used in SQLAlchemy session queries.

DEFAULT_PER_PAGE = 50

Default per_page argument for pagination when per_page not specified.

all_entities

Return list of entities + join_entities present in query.

defer(*columns)[source]

Apply defer() to query.

entities

Return list of entity classes present in query.

immediateload(*keys, **kargs)[source]

Apply immediateload() to keys.

Parameters:keys (mixed) – Either string or column references to join path(s).
Keyword Arguments:
 options (list) – A list of LoadOption to apply to the overall load strategy, i.e., each LoadOption will be chained at the end of the load.

Note

Additional keyword args will be passed to initial load creation.

join_eager(*keys, **kargs)[source]

Apply join + self.options(contains_eager()).

Parameters:

keys (mixed) – Either string or column references to join path(s).
Keyword Arguments:
 
  • alias – Join alias or dict mapping key names to aliases.
  • options (list) – A list of LoadOption to apply to the overall load strategy, i.e., each LoadOption will be chained at the end of the load.
join_entities

Return list of the joined entity classes present in query.

joinedload(*keys, **kargs)[source]

Apply joinedload() to keys.

Parameters:keys (mixed) – Either string or column references to join path(s).
Keyword Arguments:
 options (list) – A list of LoadOption to apply to the overall load strategy, i.e., each LoadOption will be chained at the end of the load.

Note

Additional keyword args will be passed to initial load creation.

lazyload(*keys, **kargs)[source]

Apply lazyload() to keys.

Parameters:keys (mixed) – Either string or column references to join path(s).
Keyword Arguments:
 options (list) – A list of LoadOption to apply to the overall load strategy, i.e., each LoadOption will be chained at the end of the load.

Note

Additional keyword args will be passed to initial load creation.

load_only(*columns)[source]

Apply load_only() to query.

map(func)[source]

Map func to each item returned by all().

noload(*keys, **kargs)[source]

Apply noload() to keys.

Parameters:keys (mixed) – Either string or column references to join path(s).
Keyword Arguments:
 options (list) – A list of LoadOption to apply to the overall load strategy, i.e., each LoadOption will be chained at the end of the load.

Note

Additional keyword args will be passed to initial load creation.

outerjoin_eager(*keys, **kargs)[source]

Apply outerjoin + self.options(contains_eager()).

Parameters:

keys (mixed) – Either string keys or column references to join path(s).
Keyword Arguments:
 
  • alias – Join alias or dict mapping key names to aliases.
  • options (list) – A list of LoadOption to apply to the overall load strategy, i.e., each LoadOption will be chained at the end of the load.
page(page=1, per_page=None)[source]

Return query with limit and offset applied for page.

paginate(page=1, per_page=None, error_out=True)[source]

Return Pagination instance using already defined query parameters.

pluck(column)[source]

Pluck column attribute values from all() results and return as list.

reduce(func, initial=None, reverse=False)[source]

Reduce all() using func.

reduce_right(func, initial=None)[source]

Reduce reversed all() using func.

subqueryload(*keys, **kargs)[source]

Apply subqueryload() to keys.

Parameters:keys (mixed) – Either string or column references to join path(s).
Keyword Arguments:
 options (list) – A list of LoadOption to apply to the overall load strategy, i.e., each LoadOption will be chained at the end of the load.

Note

Additional keyword args will be passed to initial load creation.

undefer(*columns)[source]

Apply undefer() to query.

undefer_group(*names)[source]

Apply undefer_group() to query.

class alchy.query.QueryModel(entities, session=None)[source]

Class used for default query property class for mymanager.query, mymanager.session.query, and MyModel.query. Can be used in other libraries/implementations when creating a session:

from sqlalchemy import orm

from alchy import QueryModel
# or if not using as query property
# from alchy import Query


session = orm.scoped_session(orm.sessionmaker())
session.configure(query_cls=QueryModel)

NOTE: If you don’t plan to use the query class as a query property, then you can use the Query class instead since it won’t include features that only work within a query property context.

__search_filters__

All available search filter functions indexed by a canonical name which will be referenced in advanced/simple search. All filter functions should take a single value and return an SQLAlchemy filter expression, i.e., {key: lambda value: Model.column_name.contains(value)}

__advanced_search__

Advanced search models search by named parameters. Generally found on advanced search forms where each field maps to a specific database field that will be queried against. If defined as a list, each item should be a key from __search_filters__. The matching __search_filters__ function will be used in the query. If defined as a dict, it should have the same format as __search_filters__.

__simple_search__

Simple search models search by phrase (like Google search). Defined like __advanced_search__.

__order_by__

Default order-by to use when alchy.model.ModelBase.query used.

Model

Return primary entity model class.

advanced_filter(search_dict=None)[source]

Return the compiled advanced search filter mapped to search_dict.

get_search_filters(keys)[source]

Return __search_filters__ filtered by keys.

search(search_string=None, search_dict=None, **search_options)[source]

Perform combination of simple/advanced searching with optional limit/offset support.

simple_filter(search_terms=None)[source]

Return the compiled simple search filter mapped to search_terms.

class alchy.query.QueryProperty(session)[source]

Query property accessor which gives a model access to query capabilities via alchy.model.ModelBase.query which is equivalent to session.query(Model).

class alchy.query.Pagination(query, page, per_page, total, items)[source]

Internal helper class returned by Query.paginate(). You can also construct it from any other SQLAlchemy query object if you are working with other libraries. Additionally it is possible to pass None as query object in which case the prev and next will no longer work.

has_next = None

True if a next page exists.

has_prev = None

True if a previous page exists.

items = None

The items for the current page.

next(error_out=False)[source]

Returns a Pagination object for the next page.

next_num = None

Number of the next page.

page = None

The current page number (1 indexed).

pages = None

The total number of pages.

per_page = None

The number of items to be displayed on a page.

prev(error_out=False)[source]

Returns a Pagination object for the previous page.

prev_num = None

Number of the previous page.

query = None

The query object that was used to create this pagination object.

total = None

The total number of items matching the query.

class alchy.query.LoadOption(strategy, *args, **kargs)[source]

Chained load option to apply to a load strategy when calling Query load methods.

Example usage:

qry = (db.session.query(Product)
       .join_eager('category',
                   options=[LoadOption('noload', 'images')]))

This would result in the noload option being chained to the eager option for Product.category and is equilvalent to:

qry = (db.session.query(Product)
       .join('category')
       .options(contains_eager('category').noload('images')))

Events

Declarative ORM event decorators and event registration.

SQLAlchemy features an ORM event API but one thing that is lacking is a way to register event handlers in a declarative way inside the Model’s class definition. To bridge this gap, this module contains a collection of decorators that enable this kind of functionality.

Instead of having to write event registration like this:

from sqlalchemy import event

from project.core import Model

class User(Model):
    _id = Column(types.Integer(), primary_key=True)
    email = Column(types.String())

def set_email_listener(target, value, oldvalue, initiator):
    print 'received "set" event for target: {0}'.format(target)
    return value

def before_insert_listener(mapper, connection, target):
    print 'received "before_insert" event for target: {0}'.format(target)

event.listen(User.email, 'set', set_email_listener, retval=True)
event.listen(User, 'before_insert', before_insert_listener)

Model Events allows one to write event registration more succinctly as:

from alchy import events

from project.core import Model

class User(Model):
    _id = Column(types.Integer(), primary_key=True)
    email = Column(types.String())

    @events.on_set('email', retval=True)
    def on_set_email(target, value, oldvalue, initiator):
        print 'received set event for target: {0}'.format(target)
        return value

    @events.before_insert()
    def before_insert(mapper, connection, target):
        print ('received "before_insert" event for target: {0}'
               .format(target))

For details on each event type’s expected function signature, see SQLAlchemy’s ORM Events.

class alchy.events.on_set(attribute, **event_kargs)[source]

Event decorator for the set event.

class alchy.events.on_append(attribute, **event_kargs)[source]

Event decorator for the append event.

class alchy.events.on_remove(attribute, **event_kargs)[source]

Event decorator for the remove event.

class alchy.events.before_delete(name, attribute, listener, kargs)[source]

Event decorator for the before_delete event.

class alchy.events.before_insert(**event_kargs)[source]

Event decorator for the before_insert event.

class alchy.events.before_update(**event_kargs)[source]

Event decorator for the before_update event.

class alchy.events.before_insert_update(**event_kargs)[source]

Event decorator for the before_insert and before_update events.

class alchy.events.after_delete(**event_kargs)[source]

Event decorator for the after_delete event.

class alchy.events.after_insert(**event_kargs)[source]

Event decorator for the after_insert event.

class alchy.events.after_update(**event_kargs)[source]

Event decorator for the after_update event.

class alchy.events.after_insert_update(**event_kargs)[source]

Event decorator for the after_insert and after_update events.

class alchy.events.on_append_result(**event_kargs)[source]

Event decorator for the append_result event.

class alchy.events.on_create_instance(**event_kargs)[source]

Event decorator for the create_instance event.

class alchy.events.on_instrument_class(**event_kargs)[source]

Event decorator for the instrument_class event.

class alchy.events.before_configured(**event_kargs)[source]

Event decorator for the before_configured event.

class alchy.events.after_configured(**event_kargs)[source]

Event decorator for the after_configured event.

class alchy.events.on_mapper_configured(**event_kargs)[source]

Event decorator for the mapper_configured event.

class alchy.events.on_populate_instance(**event_kargs)[source]

Event decorator for the populate_instance event.

class alchy.events.on_translate_row(**event_kargs)[source]

Event decorator for the translate_row event.

class alchy.events.on_expire(**event_kargs)[source]

Event decorator for the expire event.

class alchy.events.on_load(**event_kargs)[source]

Event decorator for the load event.

class alchy.events.on_refresh(**event_kargs)[source]

Event decorator for the refresh event.

Search

SQLAlchemy query filter factories usable in alchy.query.QueryModel.__search_filters__.

These are factory functions that return common filter operations as functions which are then assigned to the model class’ search config attributes. These functions are syntatic sugar to make it easier to define compatible search functions. However, due to the fact that a model’s query class has to be defined before the model and given that the model column attributes need to be defined before using the search factories, there are two ways to use the search factories on the query class:

  1. Define alchy.query.QueryModel.__search_filters__ as a property that returns the filter dict.
  2. Pass in a callable that returns the column.

For example, without alchy.search one would define a alchy.query.QueryModel.__search_filters__ similar to:

class UserQuery(QueryModel):
    __search_filters = {
        'email': lambda value: User.email.like(value)
    }

class User(Model):
    query_class = UserQuery
    email = Column(types.String(100))

Using alchy.search the above then becomes:

class UserQuery(QueryModel):
    @property
    def __search_filters__(self):
        return {
            'email': like(User.email)
        }

class User(Model):
    query_class = UserQuery
    email = Column(types.String(100))

Or if a callable is passed in:

from alchy import search

class UserQuery(QueryModel):
    __search_filters__ = {
        'email': like(lambda: User.email)
    }

class User(Model):
    query_class = UserQuery
    email = Column(types.String(100))

The general naming convention for each comparator is:

  • positive comparator: <base> (e.g. like())
  • negative comparator: not<base> (e.g. notlike())

The basic call signature for the search functions is:

# search.<function>(column)
# e.g.
search.contains(email)
class alchy.search.like(column)[source]

Return like filter function using ORM column field.

class alchy.search.notlike(column)[source]

Return not(like) filter function using ORM column field.

class alchy.search.ilike(column)[source]

Return ilike filter function using ORM column field.

class alchy.search.notilike(column)[source]

Return not(ilike) filter function using ORM column field.

class alchy.search.startswith(column)[source]

Return startswith filter function using ORM column field.

class alchy.search.notstartswith(column)[source]

Return not(startswith) filter function using ORM column field.

class alchy.search.endswith(column)[source]

Return endswith filter function using ORM column field.

class alchy.search.notendswith(column)[source]

Return not(endswith) filter function using ORM column field.

class alchy.search.contains(column)[source]

Return contains filter function using ORM column field.

class alchy.search.notcontains(column)[source]

Return not(contains) filter function using ORM column field.

class alchy.search.icontains(column)[source]

Return icontains filter function using ORM column field.

class alchy.search.noticontains(column)[source]

Return not(icontains) filter function using ORM column field.

class alchy.search.in_(column)[source]

Return in_ filter function using ORM column field.

class alchy.search.notin_(column)[source]

Return not(in_) filter function using ORM column field.

class alchy.search.eq(column)[source]

Return == filter function using ORM column field.

class alchy.search.noteq(column)[source]

Return not(==) filter function using ORM column field.

class alchy.search.gt(column)[source]

Return > filter function using ORM column field.

class alchy.search.notgt(column)[source]

Return not(>) filter function using ORM column field.

class alchy.search.ge(column)[source]

Return >= filter function using ORM column field.

class alchy.search.notge(column)[source]

Return not(>=) filter function using ORM column field.

class alchy.search.lt(column)[source]

Return < filter function using ORM column field.

class alchy.search.notlt(column)[source]

Return not(<) filter function using ORM column field.

class alchy.search.le(column)[source]

Return <= filter function using ORM column field.

class alchy.search.notle(column)[source]

Return not(<=) filter function using ORM column field.

class alchy.search.any_(column, column_operator)[source]

Return any filter function using ORM relationship field.

class alchy.search.notany_(column, column_operator)[source]

Return not(any) filter function using ORM relationship field.

class alchy.search.has(column, column_operator)[source]

Return has filter function using ORM relationship field.

class alchy.search.nothas(column, column_operator)[source]

Return not(has) filter function using ORM relationship field.

class alchy.search.eqenum(column, enum_class)[source]

Return == filter function using ORM DeclarativeEnum field.

class alchy.search.noteqenum(column, enum_class)[source]

Return not(==) filter function using ORM DeclarativeEnum field.

Types

Collection of custom column types.

class alchy.types.DeclarativeEnumType(enum, name=None)[source]

Column type usable in table column definitions.

class alchy.types.DeclarativeEnum[source]

Declarative enumeration.

For example:

class OrderStatus(DeclarativeEnum):
    pending = ('p', 'Pending')
    submitted = ('s', 'Submitted')
    complete = ('c', 'Complete')

class Order(Model):
    status = Column(OrderStatus.db_type(), default=OrderStatus.pending)
classmethod db_type(name=None)[source]

Return database column type for use in table column definitions.

classmethod from_string(string)[source]

Return enum symbol given string value.

Raises:ValueError – If string doesn’t correspond to an enum value.
classmethod values()[source]

Return list of possible enum values. Each value is a valid argument to from_string().

Manager

Manager class and mixin.

The Manager class helps manage a SQLAlchemy database session as well as provide convenience functions for commons operations.

Configuration

The following configuration values can be passed into a new Manager instance as a dict, class, or module.

SQLALCHEMY_DATABASE_URI URI used to connect to the database. Defaults to sqlite://.
SQLALCHEMY_BINDS A dict that maps bind keys to database URIs. Optionally, in place of a database URI, a configuration dict can be used to overrided connection options.
SQLALCHEMY_ECHO When True have SQLAlchemy echo all SQL statements. Defaults to False.
SQLALCHEMY_POOL_SIZE The size of the database pool. Defaults to the engine’s default (usually 5).
SQLALCHEMY_POOL_TIMEOUT Specifies the connection timeout for the pool. Defaults to 10.
SQLALCHEMY_POOL_RECYCLE Number of seconds after which a connection is automatically recycled.
SQLALCHEMY_MAX_OVERFLOW Controls the number of connections that can be created after the pool reached its maximum size. When those additional connections are returned to the pool, they are disconnected and discarded.
class alchy.manager.ManagerMixin[source]

Extensions for Manager.session.

add(*instances)[source]

Override session.add() so it can function like session.add_all().

Note

Supports chaining.

add_commit(*instances)[source]

Add instances to session and commit in one call.

delete(*instances)[source]

Override session.delete() so it can function like session.add_all().

Note

Supports chaining.

delete_commit(*instances)[source]

Delete instances to session and commit in one call.

class alchy.manager.Manager(config=None, session_options=None, Model=None, session_class=None)[source]

Manager class for database session.

Initialization of Manager accepts a config object, session options, and an optional declarative base. If Model isn’t provided, then a default one is generated using alchy.model.make_declarative_base(). The declarative base model is accessible at Model.

By default the session_options are:

{
    'query_cls': alchy.Query,
    'autocommit': False,
    'autoflush': True
}

The default session_class is alchy.Session. If you want to provide your own session class, then it’s suggested that you subclass alchy.Session and pass it in via session_class. This way your subclass will inherit the functionality of alchy.Session.

Model = None

Declarative base model class.

__getattr__(attr)[source]

Delegate all other attributes to session.

binds

Returns config options for all binds.

binds_map

Returns a dictionary with a table->engine mapping. This is suitable for use in sessionmaker(binds=binds_map).

config = None

Database engine configuration options.

create_all(bind='__all__')[source]

Create database schema from models.

create_engine(uri_or_config)[source]

Create engine using either a URI or a config dict. If URI supplied, then the default config will be used. If config supplied, then URI in config will be used.

create_scoped_session(options=None)[source]

Create scoped session which internally calls create_session().

create_session(options)[source]

Create session instance using custom Session class that supports multiple bindings.

drop_all(bind='__all__')[source]

Drop tables defined by models.

engine

Return default database engine.

get_engine(bind=None)[source]

Return engine associated with bind. Create engine if it doesn’t already exist.

get_tables_for_bind(bind=None)[source]

Returns a list of all tables relevant for a bind.

metadata

Return Model metadata object.

reflect(bind='__all__')[source]

Reflect tables from database.

session = None

Scoped session object.

session_class = None

Class to used for session object.

class alchy.manager.Config(defaults=None)[source]

Configuration loader which acts like a dict but supports loading values from an object limited to ALL_CAPS_ATTRIBUTES.

from_object(obj)[source]

Pull dir(obj) keys from obj and set onto self.

Session

Session class that supports multiple database binds.

class alchy.session.Session(manager, **options)[source]

The default session used by alchy.manager.Manager. It extends the default session system with bind selection.

get_bind(mapper, clause=None)[source]

Return engine bind using mapper info’s bind_key if present.

Utils

Generic utility functions used in package.

alchy.utils.is_sequence(obj)[source]

Test if obj is an iterable but not dict or str. Mainly used to determine if obj can be treated like a list for iteration purposes.

alchy.utils.has_primary_key(metadict)[source]

Check if meta class’ dict object has a primary key defined.

alchy.utils.camelcase_to_underscore(string)[source]

Convert string from CamelCase to under_score.

alchy.utils.iterflatten(items)[source]

Return iterator which flattens list/tuple of lists/tuples:

>>> to_flatten = [1, [2,3], [4, [5, [6]], 7], 8]
>>> assert list(iterflatten(to_flatten)) == [1,2,3,4,5,6,7,8]
alchy.utils.flatten(items)[source]

Return flattened list of a list/tuple of lists/tuples:

>>> assert flatten([1, [2,3], [4, [5, [6]], 7], 8]) == [1,2,3,4,5,6,7,8]