alchy¶
A SQLAlchemy extension for its declarative ORM that provides enhancements for model classes, queries, and sessions.
MAINTENANCE MODE¶
PROJECT IS IN MAINTENANCE MODE: NO NEW FEATURES, BUG FIXES ONLY
Use sqlservice instead.
Links¶
- Project: https://github.com/dgilland/alchy
- Documentation: https://alchy.readthedocs.io
- PyPi: https://pypi.python.org/pypi/alchy/
- TravisCI: https://travis-ci.org/dgilland/alchy
Quickstart¶
Let’s see alchy in action. We’ll start with some model definitions.
from alchy import ModelBase, make_declarative_base
from sqlalchemy import orm, Column, types, ForeignKey
class Base(ModelBase):
# extend/override ModelBase if necessary
pass
Model = make_declarative_base(Base=Base)
class User(Model):
__tablename__ = 'user'
_id = Column(types.Integer(), primary_key=True)
name = Column(types.String())
email = Column(types.String())
level = Column(types.Integer())
items = orm.relationship('UserItem')
class UserItem(Model):
# when no __tablename__ defined,
# one is autogenerated using class name
# like this:
#__tablename__ = 'user_item'
_id = Column(types.Integer(), primary_key=True)
user_id = Column(types.Integer(), ForeignKey('user._id'))
name = Column(types.String())
user = orm.relationship('User')
Next, we need to interact with our database. For that we will use a alchy.manager.Manager.
from alchy import Manager
# Config can be either (1) dict, (2) class, or (3) module.
config = {
'SQLALCHEMY_DATABASE_URI': 'sqlite://'
}
# Be sure to pass in our declarative base defined previously.
# This is needed so that Model.metadata operations like
# create_all(), drop_all(), and reflect() work.
db = Manager(config=config, Model=Model)
Create our database tables.
db.create_all()
Now, create some records.
# initialize using keyword args
user1 = User(name='Fred', email='fred@example.com')
# print('user1:', user1)
# ...or initialize using a dict
user2 = User({'name': 'Barney'})
# print('user2:', user2)
# update using either method as well
user2.update(email='barney@example.org')
user2.update({'email': 'barney@example.com'})
# print('user2 updated:', user2)
Add them to the database.
# there are several options for adding records
# add and commit in one step using positional args
db.add_commit(user1, user2)
# ...or add/commit using a list
users = [user1, user2]
db.add_commit(users)
# ...or separate add and commit calls
db.add(user1, user2)
db.commit()
# ...or with a list
db.add(users)
db.commit()
# ...or separate adds and commit
db.add(user1)
db.add(user2)
db.commit()
Fetch model and operate.
# create user
db.add_commit(User(name='Wilma', email='wilma@example.com'))
# fetch from database
user = User.get(user1._id)
# print('user:', user)
# convert to dict
user_dict = user.to_dict()
# print('user dict:', user_dict)
# ...or just pass object directly to dict()
user_dict = dict(user)
# make some changes
user.update(level=5)
# and refresh
user.refresh()
# or flush
user.flush()
# access the session that loaded the model instance
assert user.session() == db.object_session(user)
# delete user
user.delete()
db.commit()
# ...or via db
db.delete(user)
db.commit()
# ...or all-in-one step
db.delete_commit(user)
Query records from the database.
# add some more users
db.add_commit(
User(items=[UserItem()]),
User(items=[UserItem()]),
User(items=[UserItem()]),
User(items=[UserItem()]),
User(items=[UserItem()])
)
# there are several syntax options for querying records
# using db.session directly
# print('all users:', db.session.query(User).all())
# ...or using db directly (i.e. db.session proxy)
assert db.query(User).all() == db.session.query(User).all()
# ...or via query property on model class
assert User.query.all() == db.session.query(User).all()
Use features from the enhanced query class.
q = User.query.join(UserItem)
# entities
assert q.entities == [User]
assert q.join_entities == [UserItem]
assert q.all_entities == [User, UserItem]
# paging
assert str(q.page(2, per_page=2)) == str(q.limit(2).offset((2-1) * 2))
# pagination
page2 = q.paginate(2, per_page=2)
assert str(page2.query) == str(q)
assert page2.page == 2
assert page2.per_page == 2
assert page2.total == q.count()
assert page2.items == q.limit(2).offset((2-1) * 2).all()
assert page2.prev_num == 1
assert page2.has_prev == True
assert page2.next_num == 3
assert page2.has_next == True
page_1 = page2.prev()
page_3 = page2.next()
# searching
# ...extend class definitions to support advanced and simple searching
User.__advanced_search__ = User.__simple_search__ = {
'user_email': lambda value: User.email.like('%{0}%'.format(value)),
'user_name': lambda value: User.name.like('%{0}%'.format(value))
}
UserItem.__advanced_search__ = {
'item_name': lambda value: UserItem.name.like('%{0}%'.format(value))
}
search = User.query.search('example.com', {'user_name': 'wilma'})
# print('search:', str(search))
assert search.count() > 0
# entity loading
User.query.join_eager(User.items)
User.query.joinedload(User.items)
User.query.lazyload(User.items)
User.query.immediateload(User.items)
User.query.noload(User.items)
User.query.subqueryload(User.items)
# column loading
User.query.load_only('_id', 'name')
User.query.defer('email')
User.query.undefer('email') # if User.email undeferred in class definition
User.query.undefer_group('group1', 'group2') # if under groups defined in class
# utilities
User.query.map(lambda user: user.level)
User.query.pluck('level')
User.query.index_by('email')
User.query.chain().value()
User.query.reduce(
lambda result, user: result + 1 if user.level > 5 else result,
initial=0
)
For more details regarding the chaining API (i.e. Query.chain()), see the pydash documentation.
Utilize ORM events.
from alchy import events
class User(Model):
__table_args__ = {
# this is needed since we're replacing the ``User`` class defined above
'extend_existing': True
}
_id = Column(types.Integer(), primary_key=True)
name = Column(types.String())
email = Column(types.String())
level = Column(types.Integer())
@events.before_insert_update()
def validate(self, *args, **kargs):
'''Validate model instance'''
# do validation
return
@events.on_set('email')
def on_set_email(self, value, oldvalue, initator):
if self.query.filter(User.email==value, User._id!=self._id).count() > 0:
raise ValueError('Email already exists in database')
user = User(email='one@example.com')
db.add_commit(user)
try:
User(email=user.email)
except ValueError as ex:
pass
Finally, clean up after ourselves.
db.drop_all()
See also
For further details consult API Reference.
Guide¶
Installation¶
To install from PyPi:
pip install alchy
Compatibility¶
- Python 2.6
- Python 2.7
- Python 3.2
- Python 3.3
- Python 3.4
Dependencies¶
- SQLAlchemy >= 0.9.0
Quickstart¶
Let’s see alchy in action. We’ll start with some model definitions.
from alchy import ModelBase, make_declarative_base
from sqlalchemy import orm, Column, types, ForeignKey
class Base(ModelBase):
# extend/override ModelBase if necessary
pass
Model = make_declarative_base(Base=Base)
class User(Model):
__tablename__ = 'user'
_id = Column(types.Integer(), primary_key=True)
name = Column(types.String())
email = Column(types.String())
level = Column(types.Integer())
items = orm.relationship('UserItem')
class UserItem(Model):
# when no __tablename__ defined,
# one is autogenerated using class name
# like this:
#__tablename__ = 'user_item'
_id = Column(types.Integer(), primary_key=True)
user_id = Column(types.Integer(), ForeignKey('user._id'))
name = Column(types.String())
user = orm.relationship('User')
Next, we need to interact with our database. For that we will use a alchy.manager.Manager.
from alchy import Manager
# Config can be either (1) dict, (2) class, or (3) module.
config = {
'SQLALCHEMY_DATABASE_URI': 'sqlite://'
}
# Be sure to pass in our declarative base defined previously.
# This is needed so that Model.metadata operations like
# create_all(), drop_all(), and reflect() work.
db = Manager(config=config, Model=Model)
Create our database tables.
db.create_all()
Now, create some records.
# initialize using keyword args
user1 = User(name='Fred', email='fred@example.com')
# print('user1:', user1)
# ...or initialize using a dict
user2 = User({'name': 'Barney'})
# print('user2:', user2)
# update using either method as well
user2.update(email='barney@example.org')
user2.update({'email': 'barney@example.com'})
# print('user2 updated:', user2)
Add them to the database.
# there are several options for adding records
# add and commit in one step using positional args
db.add_commit(user1, user2)
# ...or add/commit using a list
users = [user1, user2]
db.add_commit(users)
# ...or separate add and commit calls
db.add(user1, user2)
db.commit()
# ...or with a list
db.add(users)
db.commit()
# ...or separate adds and commit
db.add(user1)
db.add(user2)
db.commit()
Fetch model and operate.
# create user
db.add_commit(User(name='Wilma', email='wilma@example.com'))
# fetch from database
user = User.get(user1._id)
# print('user:', user)
# convert to dict
user_dict = user.to_dict()
# print('user dict:', user_dict)
# ...or just pass object directly to dict()
user_dict = dict(user)
# make some changes
user.update(level=5)
# and refresh
user.refresh()
# or flush
user.flush()
# access the session that loaded the model instance
assert user.session() == db.object_session(user)
# delete user
user.delete()
db.commit()
# ...or via db
db.delete(user)
db.commit()
# ...or all-in-one step
db.delete_commit(user)
Query records from the database.
# add some more users
db.add_commit(
User(items=[UserItem()]),
User(items=[UserItem()]),
User(items=[UserItem()]),
User(items=[UserItem()]),
User(items=[UserItem()])
)
# there are several syntax options for querying records
# using db.session directly
# print('all users:', db.session.query(User).all())
# ...or using db directly (i.e. db.session proxy)
assert db.query(User).all() == db.session.query(User).all()
# ...or via query property on model class
assert User.query.all() == db.session.query(User).all()
Use features from the enhanced query class.
q = User.query.join(UserItem)
# entities
assert q.entities == [User]
assert q.join_entities == [UserItem]
assert q.all_entities == [User, UserItem]
# paging
assert str(q.page(2, per_page=2)) == str(q.limit(2).offset((2-1) * 2))
# pagination
page2 = q.paginate(2, per_page=2)
assert str(page2.query) == str(q)
assert page2.page == 2
assert page2.per_page == 2
assert page2.total == q.count()
assert page2.items == q.limit(2).offset((2-1) * 2).all()
assert page2.prev_num == 1
assert page2.has_prev == True
assert page2.next_num == 3
assert page2.has_next == True
page_1 = page2.prev()
page_3 = page2.next()
# searching
# ...extend class definitions to support advanced and simple searching
User.__advanced_search__ = User.__simple_search__ = {
'user_email': lambda value: User.email.like('%{0}%'.format(value)),
'user_name': lambda value: User.name.like('%{0}%'.format(value))
}
UserItem.__advanced_search__ = {
'item_name': lambda value: UserItem.name.like('%{0}%'.format(value))
}
search = User.query.search('example.com', {'user_name': 'wilma'})
# print('search:', str(search))
assert search.count() > 0
# entity loading
User.query.join_eager(User.items)
User.query.joinedload(User.items)
User.query.lazyload(User.items)
User.query.immediateload(User.items)
User.query.noload(User.items)
User.query.subqueryload(User.items)
# column loading
User.query.load_only('_id', 'name')
User.query.defer('email')
User.query.undefer('email') # if User.email undeferred in class definition
User.query.undefer_group('group1', 'group2') # if under groups defined in class
# utilities
User.query.map(lambda user: user.level)
User.query.pluck('level')
User.query.index_by('email')
User.query.chain().value()
User.query.reduce(
lambda result, user: result + 1 if user.level > 5 else result,
initial=0
)
For more details regarding the chaining API (i.e. Query.chain()), see the pydash documentation.
Utilize ORM events.
from alchy import events
class User(Model):
__table_args__ = {
# this is needed since we're replacing the ``User`` class defined above
'extend_existing': True
}
_id = Column(types.Integer(), primary_key=True)
name = Column(types.String())
email = Column(types.String())
level = Column(types.Integer())
@events.before_insert_update()
def validate(self, *args, **kargs):
'''Validate model instance'''
# do validation
return
@events.on_set('email')
def on_set_email(self, value, oldvalue, initator):
if self.query.filter(User.email==value, User._id!=self._id).count() > 0:
raise ValueError('Email already exists in database')
user = User(email='one@example.com')
db.add_commit(user)
try:
User(email=user.email)
except ValueError as ex:
pass
Finally, clean up after ourselves.
db.drop_all()
See also
For further details consult API Reference.
Versioning¶
This project follows Semantic Versioning.
Upgrading¶
To v2.0.0¶
The logic that sets a Model class’ __table_args__ and __mapper_args__ (unless overridden in subclass) has been modified. A model’s __local_table_args__ and __local_mapper_args__ are now merged with __global_table_args__ and __global_mapper_args__ from all classes in the class’s mro(). A __{global,local}_{table,mapper}_args__ may be a callable or classmethod, in which case it is evaluated on the class whose __{table,mapper}_args__ is being set.
To v1.0.0¶
The @classproperty decorator has been eliminated and replaced with @classmethod in v1.0.0. This means that the previous alchy.model.ModelBase class properties must now be accessed via method calls:
alchy.model.ModelBase.session()alchy.model.ModelBase.primary_key()alchy.model.ModelBase.primary_keys()alchy.model.ModelBase.primary_attrs()alchy.model.ModelBase.attrs()alchy.model.ModelBase.descriptors()alchy.model.ModelBase.relationships()alchy.model.ModelBase.column_attrs()alchy.model.ModelBase.columns()
API Reference¶
Includes links to source code.
API Reference¶
Model¶
Declarative base for ORM models.
-
class
alchy.model.ModelMeta(name, bases, dct)[source]¶ ModelBase’s metaclass which provides:
- Tablename autogeneration when
__tablename__not defined. - Support for multiple database bindings via
ModelBase.__bind_key__. - Support for declarative ORM events via
alchy.eventsdecorators orModelBase.__events__.
- Tablename autogeneration when
-
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 toManager.config['SQLALCHEMY_DATABASE_URI'].
-
__events__¶ Register orm event listeners. See
alchy.eventsfor 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 settingMyClass.query = QueryProperty(session)when session available. Seemake_declarative_base()for automatic implementation.
-
__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 callingto_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 callingrefresh()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
column_attrs()[source]¶ Return table columns as list of class attributes at the class level.
-
descriptor_dict¶ Return
__dict__key-filtered bydescriptors.
-
classmethod
get_by(data_dict=None, **kargs)[source]¶ Return first instance filtered by values using
cls.query.filter_by().
-
object_session¶ Return session belonging to
self
-
classmethod
primary_key()[source]¶ Return primary key as either single column (one primary key) or tuple otherwise.
-
query_class alias of
QueryModel
-
to_dict()[source]¶ Return dict representation of model by filtering fields using
__to_dict__.
-
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.
-
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 LoadOptionto apply to the overall load strategy, i.e., eachLoadOptionwill 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
dictmapping key names to aliases. - options (list) –
A list of
LoadOptionto apply to the overall load strategy, i.e., eachLoadOptionwill be chained at the end of the load.
- alias –
Join alias or
-
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 LoadOptionto apply to the overall load strategy, i.e., eachLoadOptionwill 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 LoadOptionto apply to the overall load strategy, i.e., eachLoadOptionwill be chained at the end of the load.Note
Additional keyword args will be passed to initial load creation.
-
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 LoadOptionto apply to the overall load strategy, i.e., eachLoadOptionwill 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
dictmapping key names to aliases. - options (list) –
A list of
LoadOptionto apply to the overall load strategy, i.e., eachLoadOptionwill be chained at the end of the load.
- alias –
Join alias or
-
paginate(page=1, per_page=None, error_out=True)[source]¶ Return
Paginationinstance using already defined query parameters.
-
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 LoadOptionto apply to the overall load strategy, i.e., eachLoadOptionwill be chained at the end of the load.Note
Additional keyword args will be passed to initial load creation.
-
-
class
alchy.query.QueryModel(entities, session=None)[source]¶ Class used for default query property class for
mymanager.query,mymanager.session.query, andMyModel.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
Queryclass 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.queryused.
-
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.
-
-
class
alchy.query.QueryProperty(session)[source]¶ Query property accessor which gives a model access to query capabilities via
alchy.model.ModelBase.querywhich is equivalent tosession.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 passNoneas 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
Paginationobject 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
Paginationobject 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
Queryload methods.Example usage:
qry = (db.session.query(Product) .join_eager('category', options=[LoadOption('noload', 'images')]))
This would result in the
noloadoption being chained to the eager option forProduct.categoryand 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_append(attribute, **event_kargs)[source]¶ Event decorator for the
appendevent.
-
class
alchy.events.on_remove(attribute, **event_kargs)[source]¶ Event decorator for the
removeevent.
-
class
alchy.events.before_delete(**event_kargs)[source]¶ Event decorator for the
before_deleteevent.
-
class
alchy.events.before_insert(**event_kargs)[source]¶ Event decorator for the
before_insertevent.
-
class
alchy.events.before_update(**event_kargs)[source]¶ Event decorator for the
before_updateevent.
-
class
alchy.events.before_insert_update(**event_kargs)[source]¶ Event decorator for the
before_insertandbefore_updateevents.
-
class
alchy.events.after_insert_update(**event_kargs)[source]¶ Event decorator for the
after_insertandafter_updateevents.
-
class
alchy.events.on_append_result(**event_kargs)[source]¶ Event decorator for the
append_resultevent.
-
class
alchy.events.on_create_instance(**event_kargs)[source]¶ Event decorator for the
create_instanceevent.
-
class
alchy.events.on_instrument_class(**event_kargs)[source]¶ Event decorator for the
instrument_classevent.
-
class
alchy.events.before_configured(**event_kargs)[source]¶ Event decorator for the
before_configuredevent.
-
class
alchy.events.after_configured(**event_kargs)[source]¶ Event decorator for the
after_configuredevent.
-
class
alchy.events.on_mapper_configured(**event_kargs)[source]¶ Event decorator for the
mapper_configuredevent.
-
class
alchy.events.on_populate_instance(**event_kargs)[source]¶ Event decorator for the
populate_instanceevent.
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:
- Define
alchy.query.QueryModel.__search_filters__as a property that returns the filter dict. - 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:
The basic call signature for the search functions is:
# search.<function>(column)
# e.g.
search.contains(email)
-
class
alchy.search.notlike(column)[source]¶ Return
not(like)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
startswithfilter 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
endswithfilter 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
containsfilter 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
icontainsfilter function using ORM column field.
-
class
alchy.search.noticontains(column)[source]¶ Return
not(icontains)filter function using ORM column field.
-
class
alchy.search.any_(column, column_operator)[source]¶ Return
anyfilter 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
hasfilter function using ORM relationship field.
-
class
alchy.search.nothas(column, column_operator)[source]¶ Return
not(has)filter function using ORM relationship 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().
-
classmethod
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 likesession.add_all().Note
Supports chaining.
-
-
class
alchy.manager.Manager(config=None, session_options=None, Model=None, session_class=None)[source]¶ Manager class for database session.
Initialization of
Manageraccepts a config object, session options, and an optional declarative base. IfModelisn’t provided, then a default one is generated usingalchy.model.make_declarative_base(). The declarative base model is accessible atModel.By default the
session_optionsare:{ 'query_cls': alchy.Query, 'autocommit': False, 'autoflush': True }
The default
session_classisalchy.Session. If you want to provide your own session class, then it’s suggested that you subclassalchy.Sessionand pass it in viasession_class. This way your subclass will inherit the functionality ofalchy.Session.-
Model= None¶ Declarative base model class.
-
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_engine(uri_or_config)[source]¶ Create engine using either a URI or a config dict. If URI supplied, then the default
configwill 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.
-
engine¶ Return default database engine.
-
get_engine(bind=None)[source]¶ Return engine associated with bind. Create engine if it doesn’t already exist.
-
session= None¶ Scoped session object.
-
session_class= None¶ Class to used for session object.
-
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.Parameters: - manager (alchy.manager.Manager) – Alchy manager instance
- options (dict) – pass-through to SessionBase call
Utils¶
Generic utility functions used in package.
-
alchy.utils.is_sequence(obj)[source]¶ Test if obj is an iterable but not
dictorstr. Mainly used to determine if obj can be treated like alistfor iteration purposes.
Project Info¶
License¶
Copyright (c) 2014 Derrick Gilland
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
Changelog¶
v2.2.2 (2017-01-03)¶
- Fix bug in handling of session options when providing explicit
bindsvalue viasession_optionsduringManagerin initialization. Thanks brianbruggeman!
v2.2.1 (2016-05-18)¶
- Fix bug with
events.before_deletewhere decorator was defined using invalid parent class making it completely non-functional as a decorator.
v2.2.0 (2016-03-21)¶
- Add
metadata``argument to ``alchy.model.make_declarative_baseto provide custom metaclass for declarative base model. Thanks fabioramponi!
v2.1.0 (2016-03-11)¶
- Add
Metaargument toalchy.model.make_declarative_baseto provide custom metaclass for declarative base model. Thanks elidchan!
v2.0.1 (2015-07-29)¶
- Make
Session.get_bind()mapperargument have default value ofNone.
v2.0.0 (2015-04-29)¶
- Add
Query.index_by. - Add
Query.chain. - Add
pydashas dependency and incorporate into existingQuerymethods:map,reduce,reduce_right, andpluck. - Improve logic for setting
__tablename__to work with all table inheritance styles (joined, single, and concrete), to handle@declared_attrcolumns, and not to duplicate underscore characters. Thanks sethp! - Modify logic that sets a Model class’
__table_args__and__mapper_args__(unless overridden in subclass) by merging__global_table_args__and__global_mapper_args__from all classes in the class’smro()with__local_table_args__and__local_mapper_args__from the class itself. A__{global,local}_{table,mapper}_args__may be callable or classmethod, in which case it is evaluated on the class whose__{table,mapper}_args__is being set. Thanks sethp! (breaking change)
v1.5.1 (2015-01-13)¶
- Add support for callable
__table_args__and__local_table_args__. Thanks sethp!
v1.5.0 (2014-12-16)¶
- Add
Model.is_modified(). Thanks sethp! - Add
Model.filter(). - Add
Model.filter_by().
v1.4.2 (2014-11-18)¶
- Add
search.inenumandsearch.notinenumfor performing anin_andnot(in_)comparision usingDeclarativeEnum.
v1.4.1 (2014-11-17)¶
- Allow
Model.__bind_key__to be set at the declarative base level so that model classes can properly inherit it.
v1.4.0 (2014-11-09)¶
v1.3.1 (2014-10-14)¶
- During
Model.update()when setting a non-list relationship automatically instantiatedictvalues using the relationship model class.
v1.3.0 (2014-10-10)¶
- Convert null relationships to
{}when callingModel.to_dict()instead of leaving asNone.
v1.2.0 (2014-10-10)¶
- During
Model.update()when setting a list relationship automatically instantiatedictvalues using the relationship model class.
v1.1.2 (2014-09-25)¶
- Allow
aliaskeyword argument toQuery.join_eager()andQuery.outerjoin_eager()to be adictmapping aliases to join keys. Enables nested aliases.
v1.1.1 (2014-09-01)¶
- Fix handling of nested
Model.update()calls to relationship attributes so that setting relationship to emptydictwill propagateNoneto relationship attribute value correctly.
v1.1.0 (2014-08-30)¶
- Add
query.LoadOptionto support nesting load options when calling thequery.Queryload methods:join_eager,outerjoin_eager,joinedload,immediateload,lazyload,noload, andsubqueryload.
v1.0.0 (2014-08-25)¶
- Replace usage of
@classpropertydecorators inModelBasewith@classmethod. Any previously defined class properties now require method access. Affected attributes are:session,primary_key,primary_keys,primary_attrs,attrs,descriptors,relationships,column_attrs, andcolumns. (breaking change) - Proxy
getitemandsetitemaccess togetattrandsetattrinModelBase. Allows models to be accessed like dictionaries. - Make
alchy.eventsdecorators class based. - Require
alchy.eventsdecorators to be instantiated using a function call (e.g.@events.before_update()instead of@events.before_update). (breaking change) - Add
alchy.searchcomparators,eqenumandnoteqenum, for comparingDeclarativeEnumtypes.
v0.13.3 (2014-07-26)¶
- Fix
utils.iterflatten()by callingiterflatten()instead offlattenin recursive loop.
v0.13.2 (2014-06-12)¶
- Add
ModelBase.primary_attrsclass property that returns a list of class attributes that are primary keys. - Use
ModelBase.primary_attrsinQueryModel.search()so that it handles cases where primary keys have column names that are different than the class attribute name.
v0.13.1 (2014-06-11)¶
- Modify internals of
QueryModel.search()to better handle searching on a query object that already has joins and filters applied.
v0.13.0 (2014-06-03)¶
- Add
search.icontainsandsearch.noticontainsfor case insensitive contains filter. - Remove strict update support from
Model.update(). Require this to be implemented in user-land. (breaking change)
v0.12.0 (2014-05-18)¶
- Merge originating query where clause in
Query.searchso that pagination works properly. - Add
session_classargument toManagerwhich can override the default session class used.
v0.11.3 (2014-05-05)¶
- In
ModelMetawhen checking whether to do tablename autogeneration, tranverse all base classes when trying to determine if a primary key is defined. - In
ModelMetasetbind_keyin__init__method instead of__new__. This also fixes an issue where__table_args__was incorrectly assumed to always be adict.
v0.11.2 (2014-05-05)¶
- Support
order_byas list/tuple inQueryModel.search().
v0.11.1 (2014-05-05)¶
- Fix bug in
QueryModel.search()whereorder_bywasn’t applied in the correct order. Needed to come before limit/offset are applied.
v0.11.0 (2014-05-04)¶
- PEP8 compliance with default settings.
- Remove
query_propertyargument frommake_declarative_base()andextend_declarative_base(). (breaking change) - Add
ModelBase.primary_keysclass property which returns a tuple always (ModelBase.primary_keyreturns a single key if only one present or a tuple if multiple). - Move location of class
QueryPropertyfromalchy.modeltoalchy.query. (breaking change) - Create new
Querysubclass namedQueryModelwhich is to be used within a query property context. ReplaceQuerywithQueryModelas default query class. (breaking change) - Move
__advanced_search__and__simple_search__class attributes fromModelBasetoQueryModel. (breaking change) - Introduce
QueryModel.__search_filters__which can define a canonical set of search filters which can then be referenced in the list version of__advanced_search__and__simple_search__. - Modify the logic of
QueryModel.search()to use a subquery joined onto the originating query in order to support pagination when one-to-many and many-to-many joins are present on the originating query. (breaking change) - Support passing in a callable that returns a column attribute for
alchy.search.<method>(). Allows foralchy.search.contains(lambda: Foo.id)to be used at the class attribute level whenFoo.idwill be defined later. - Add search operators
any_/notany_andhas/nothaswhich can be used for the corresponding relationship operators.
v0.10.0 (2014-04-02)¶
- Issue warning instead of failing when installed version of SQLAlchemy isn’t compatible with
alchy.Query‘s loading API (i.e. missingsqlalchemy.orm.strategy_options.Load). This allowsalchyto be used with earlier versions of SQLAlchemy at user’s own risk. - Add
alchy.searchmodule which provides compatible search functions forModelBase.__advanced_search__andModelBase.__simple_search__.
v0.9.1 (2014-03-30)¶
- Change
ModelBase.sessionto proxyModelBase.query.session. - Add
ModelBase.object_sessionproxy toorm.object_session(ModelBase).
v0.9.0 (2014-03-26)¶
- Remove
engine_config_prefixargument toManager(). (breaking change) - Add explicit
session_optionsargument toManager(). (breaking change) - Change the
Manager.configoptions to follow Flask-SQLAlchemy. (breaking change) - Allow
Manager.configto be either adict,class, ormodule object. - Add multiple database engine support using a single
Managerinstance. - Add
__bind_key__configuration option forModelBasefor binding model to specific database bind (similar to Flask-SQLAlchemy).
v0.8.0 (2014-03-18)¶
- For
ModelBase.update()don’t nestupdate()calls if field attribute is adict. - Deprecated
refresh_on_emptyargument toModelBase.to_dict()and instead implementModelBase.__to_dict__configuration property as place to handle processing of model before casting todict. (breaking change) - Add
ModelBase.__to_dict__configuration property which handles preprocessing for model instance and returns a set of fields as strings to be used as dict keys when callingto_dict().
v0.7.0 (2014-03-13)¶
- Rename
alchy.ManagerBasetoalchy.ManagerMixin. (breaking change) - Add
pylintsupport. - Remove dependency on
six.
v0.6.0 (2014-03-10)¶
- Prefix event decorators which did not start with
before_orafter_withon_. Specifically,on_set,on_append,on_remove,on_append_result,on_create_instance,on_instrument_class,on_mapper_configured,on_populate_instance,on_translate_row,on_expire,on_load, andon_refresh. (breaking change) - Remove lazy engine/session initialization in
Manager. Require thatModelandconfigbe passed in at init time. While this removes some functionality, it’s done to simplify theManagercode so that it’s more straightforward. If lazy initialization is needed, then a proxy class should be used. (breaking change)
v0.5.0 (2014-03-02)¶
- Add
ModelBase.primary_keyclass property for retrieving primary key(s). - Add
Base=Noneargument tomake_declarative_base()to support passing in a subclass ofModelBase. Previously had to create a declarativeModelto pass in a subclassedModelBase. - Let any exception occurring in
ModelBase.queryattribute access bubble up (previously,UnmappedClassErrorwas caught). - Python 2.6 and 3.3 support.
- PEP8 compliance.
- New dependency:
six(for Python 3 support)
v0.4.2 (2014-02-24)¶
- In
ModelBase.to_dict()only include fields which are mapper descriptors. - Support
to_dictmethod hook when iterating over objects inModelBase.to_dict(). - Add
to_dictmethod hook toEnumSymbol(propagates toDeclarativeEnum).
v0.4.1 (2014-02-23)¶
- Support
__iter__method in model so thatdict(model)is equilvalent tomodel.to_dict(). - Add
refresh_on_empty=Trueargument toModelBase.to_dict()which supports callingModelBase.refresh()if__dict__is empty.
v0.4.0 (2014-02-23)¶
- Add
ModelBase.save()method which adds model instance loaded from session to transaction. - Add
ModelBase.get_by()which proxies toModelBase.query.filter_by().first(). - Add model attribute
events. - Add support for multiple event decoration.
- Add named events for all supported events.
- Add composite events for
before_insert_updateandafter_insert_update.
v0.3.0 (2014-02-07)¶
- Rename
ModelBase.advanced_search_configtoModelBase.__advanced_search__. - Rename
ModelBase.simple_search_configtoModelBase.__simple_search__ - Add
ModelMetametaclass. - Implement
__tablename__autogeneration from class name. - Add mapper event support via
ModelBase.__events__and/ormodel.eventdecorator.
v0.2.1 (2014-02-03)¶
- Fix reference to
model.make_declarative_baseinManagerclass.
v0.2.0 (2014-02-02)¶
- Add default
query_classto declarative model if none defined. - Let
model.make_declarative_base()accept predefined base and just extend its functionality.
v0.1.0 (2014-02-01)¶
- First release
Authors¶
Lead¶
- Derrick Gilland, dgilland@gmail.com, dgilland@github
Contributors¶
- Jeff Knupp, http://www.jeffknupp.com/, jeffknupp@github
- Seth P., seth-p@github
- Eli Chan, eli.d.chan@gmail.com, elidchan@github
- Brian Bruggeman, brian.m.bruggeman@gmail.com, brianbruggeman@github
Contributing¶
Contributions are welcome, and they are greatly appreciated! Every little bit helps, and credit will always be given.
You can contribute in many ways:
Types of Contributions¶
Report Bugs¶
Report bugs at https://github.com/dgilland/alchy.
If you are reporting a bug, please include:
- Your operating system name and version.
- Any details about your local setup that might be helpful in troubleshooting.
- Detailed steps to reproduce the bug.
Fix Bugs¶
Look through the GitHub issues for bugs. Anything tagged with “bug” is open to whoever wants to implement it.
Implement Features¶
Look through the GitHub issues for features. Anything tagged with “enhancement” or “help wanted” is open to whoever wants to implement it.
Write Documentation¶
Pydash could always use more documentation, whether as part of the official alchy docs, in docstrings, or even on the web in blog posts, articles, and such.
Submit Feedback¶
The best way to send feedback is to file an issue at https://github.com/dgilland/alchy.
If you are proposing a feature:
- Explain in detail how it would work.
- Keep the scope as narrow as possible, to make it easier to implement.
- Remember that this is a volunteer-driven project, and that contributions are welcome :)
Get Started!¶
Ready to contribute? Here’s how to set up alchy for local development.
Fork the
alchyrepo on GitHub.Clone your fork locally:
$ git clone git@github.com:your_name_here/alchy.git
Install your local copy into a virtualenv. Assuming you have virtualenv installed, this is how you set up your fork for local development:
$ cd alchy $ pip install -r requirements-dev.txt
Create a branch for local development:
$ git checkout -b name-of-your-bugfix-or-feature
Now you can make your changes locally.
When you’re done making changes, check that your changes pass linting and all unit tests by testing with tox across all supported Python versions:
$ invoke tox
Add yourself to
AUTHORS.rst.Commit your changes and push your branch to GitHub:
$ git add . $ git commit -m "Detailed description of your changes." $ git push origin name-of-your-bugfix-or-feature
Submit a pull request through the GitHub website.
Pull Request Guidelines¶
Before you submit a pull request, check that it meets these guidelines:
- The pull request should include tests.
- If the pull request adds functionality, the docs should be updated. Put your new functionality into a function with a docstring, and add the feature to the README.rst.
- The pull request should work for Python 2.7, 3.4, and 3.5. Check https://travis-ci.org/dgilland/alchy/pull_requests and make sure that the tests pass for all supported Python versions.
Kudos¶
Thank you to SQLAlchemy for providing such a great library to work with.