Source code for flask_unchained.bundles.sqlalchemy.sqla.events

from flask_unchained.string_utils import slugify as _slugify
from functools import partial
from sqlalchemy import event

# EVENTS DOCS
# http://docs.sqlalchemy.org/en/rel_1_1/core/event.html
# ORM EVENTS DOCS
# http://docs.sqlalchemy.org/en/rel_1_1/orm/events.html


class _SQLAlchemyEvent:
    """Private helper class for the @attach_events and @on decorators"""
    ATTR = '_sqlalchemy_event'

    def __init__(self, field_name, event_name, listen_kwargs=None):
        self.field_name = field_name
        self.event_name = event_name
        self.listen_kwargs = listen_kwargs or {}


[docs]def attach_events(*args): """Class decorator for SQLAlchemy models to attach listeners for class methods decorated with :func:`.on` Usage:: @attach_events class User(Model): email = Column(String(50)) @on('email', 'set') def lowercase_email(self, new_value, old_value, initiating_event): self.email = new_value.lower() """ def wrapper(cls): for name, fn in cls.__dict__.items(): if not name.startswith('__') and hasattr(fn, _SQLAlchemyEvent.ATTR): e = getattr(fn, _SQLAlchemyEvent.ATTR) if e.field_name: event.listen(getattr(cls, e.field_name), e.event_name, fn, **e.listen_kwargs) else: event.listen(cls, e.event_name, fn, **e.listen_kwargs) return cls if args and callable(args[0]): return wrapper(args[0]) return wrapper
[docs]def on(*args, **listen_kwargs): """Class method decorator for SQLAlchemy models. Must be used in conjunction with the :func:`.attach_events` class decorator Usage:: @attach_events class Post(Model): uuid = Column(String(36)) post_tags = relationship('PostTag', back_populates='post') # m2m # instance event (only one positional argument, the event name) # kwargs are passed on to the sqlalchemy.event.listen function @on('init', once=True) def generate_uuid(self, args, kwargs): self.uuid = str(uuid.uuid4()) # attribute event (two positional args, field name and event name) @on('post_tags', 'append') def set_tag_order(self, post_tag, initiating_event): if not post_tag.order: post_tag.order = len(self.post_tags) + 1 """ if len(args) == 1: field_name, event_name = (None, args[0]) elif len(args) == 2: field_name, event_name = args else: raise NotImplementedError('@on accepts only one or two positional arguments') def wrapper(fn): setattr(fn, _SQLAlchemyEvent.ATTR, _SQLAlchemyEvent(field_name, event_name, listen_kwargs)) return fn return wrapper
[docs]def slugify(field_name, slug_field_name=None, mutable=False): """Class decorator to specify a field to slugify. Slugs are immutable by default unless mutable=True is passed. Usage:: @slugify('title') def Post(Model): title = Column(String(100)) slug = Column(String(100)) # pass a second argument to specify the slug attribute field: @slugify('title', 'title_slug') def Post(Model) title = Column(String(100)) title_slug = Column(String(100)) # optionally set mutable to True for a slug that changes every time # the slugified field changes: @slugify('title', mutable=True) def Post(Model): title = Column(String(100)) slug = Column(String(100)) """ slug_field_name = slug_field_name or 'slug' def _set_slug(target, value, old_value, _, mutable=False): existing_slug = getattr(target, slug_field_name) if existing_slug and not mutable: return if value and (not existing_slug or value != old_value): setattr(target, slug_field_name, _slugify(value)) def wrapper(cls): event.listen(getattr(cls, field_name), 'set', partial(_set_slug, mutable=mutable)) return cls return wrapper