Listeners

Automatic data coercion

sqlalchemy_utils.listeners.force_auto_coercion(mapper=None)

Function that assigns automatic data type coercion for all classes which are of type of given mapper. The coercion is applied to all coercion capable properties. By default coercion is applied to all SQLAlchemy mappers.

Before initializing your models you need to call force_auto_coercion.

from sqlalchemy_utils import force_auto_coercion


force_auto_coercion()

Then define your models the usual way:

class Document(Base):
    __tablename__ = 'document'
    id = sa.Column(sa.Integer, autoincrement=True)
    name = sa.Column(sa.Unicode(50))
    background_color = sa.Column(ColorType)

Now scalar values for coercion capable data types will convert to appropriate value objects:

document = Document()
document.background_color = 'F5F5F5'
document.background_color  # Color object
session.commit()

Parameters:mapper – The mapper which the automatic data type coercion should be applied to

Instant defaults

sqlalchemy_utils.listeners.force_instant_defaults(mapper=None)

Function that assigns object column defaults on object initialization time. By default calling this function applies instant defaults to all your models.

Setting up instant defaults:

from sqlalchemy_utils import force_instant_defaults


force_instant_defaults()

Example usage:

class Document(Base):
    __tablename__ = 'document'
    id = sa.Column(sa.Integer, autoincrement=True)
    name = sa.Column(sa.Unicode(50))
    created_at = sa.Column(sa.DateTime, default=datetime.now)


document = Document()
document.created_at  # datetime object

Parameters:mapper – The mapper which the automatic instant defaults forcing should be applied to

Many-to-many orphan deletion

sqlalchemy_utils.listeners.auto_delete_orphans(attr)

Delete orphans for given SQLAlchemy model attribute. This function can be used for deleting many-to-many associated orphans easily. For more information see https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/ManyToManyOrphan.

Consider the following model definition:

from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event


Base = declarative_base()

tagging = Table(
    'tagging',
    Base.metadata,
    Column(
        'tag_id',
        Integer,
        ForeignKey('tag.id', ondelete='CASCADE'),
        primary_key=True
    ),
    Column(
        'entry_id',
        Integer,
        ForeignKey('entry.id', ondelete='CASCADE'),
        primary_key=True
    )
)

class Tag(Base):
    __tablename__ = 'tag'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), unique=True, nullable=False)

    def __init__(self, name=None):
        self.name = name

class Entry(Base):
    __tablename__ = 'entry'

    id = Column(Integer, primary_key=True)

    tags = relationship(
        'Tag',
        secondary=tagging,
        backref='entries'
    )

Now lets say we want to delete the tags if all their parents get deleted ( all Entry objects get deleted). This can be achieved as follows:

from sqlalchemy_utils import auto_delete_orphans


auto_delete_orphans(Entry.tags)

After we’ve set up this listener we can see it in action.

e = create_engine('sqlite://')

Base.metadata.create_all(e)

s = Session(e)

r1 = Entry()
r2 = Entry()
r3 = Entry()
t1, t2, t3, t4 = Tag('t1'), Tag('t2'), Tag('t3'), Tag('t4')

r1.tags.extend([t1, t2])
r2.tags.extend([t2, t3])
r3.tags.extend([t4])
s.add_all([r1, r2, r3])

assert s.query(Tag).count() == 4

r2.tags.remove(t2)

assert s.query(Tag).count() == 4

r1.tags.remove(t2)

assert s.query(Tag).count() == 3

r1.tags.remove(t1)

assert s.query(Tag).count() == 2

Parameters:attr – Association relationship attribute to auto delete orphans from