Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from collections import OrderedDict
2from functools import partial
3from inspect import isclass
4from operator import attrgetter
6import sqlalchemy as sa
7from sqlalchemy.engine.interfaces import Dialect
8from sqlalchemy.ext.hybrid import hybrid_property
9from sqlalchemy.orm import ColumnProperty, mapperlib, RelationshipProperty
10from sqlalchemy.orm.attributes import InstrumentedAttribute
11from sqlalchemy.orm.exc import UnmappedInstanceError
13from sqlalchemy.orm.context import _ColumnEntity, _MapperEntity
15from sqlalchemy.orm.session import object_session
16from sqlalchemy.orm.util import AliasedInsp
18from ..utils import is_sequence
21def get_class_by_table(base, table, data=None):
22 """
23 Return declarative class associated with given table. If no class is found
24 this function returns `None`. If multiple classes were found (polymorphic
25 cases) additional `data` parameter can be given to hint which class
26 to return.
28 ::
30 class User(Base):
31 __tablename__ = 'entity'
32 id = sa.Column(sa.Integer, primary_key=True)
33 name = sa.Column(sa.String)
36 get_class_by_table(Base, User.__table__) # User class
39 This function also supports models using single table inheritance.
40 Additional data paratemer should be provided in these case.
42 ::
44 class Entity(Base):
45 __tablename__ = 'entity'
46 id = sa.Column(sa.Integer, primary_key=True)
47 name = sa.Column(sa.String)
48 type = sa.Column(sa.String)
49 __mapper_args__ = {
50 'polymorphic_on': type,
51 'polymorphic_identity': 'entity'
52 }
54 class User(Entity):
55 __mapper_args__ = {
56 'polymorphic_identity': 'user'
57 }
60 # Entity class
61 get_class_by_table(Base, Entity.__table__, {'type': 'entity'})
63 # User class
64 get_class_by_table(Base, Entity.__table__, {'type': 'user'})
67 :param base: Declarative model base
68 :param table: SQLAlchemy Table object
69 :param data: Data row to determine the class in polymorphic scenarios
70 :return: Declarative class or None.
71 """
72 found_classes = {
73 c
74 for c in _get_class_registry(base).values()
75 if hasattr(c, '__table__') and c.__table__ is table
76 }
77 if len(found_classes) > 1:
78 if not data:
79 raise ValueError(
80 "Multiple declarative classes found for table '{}'. "
81 'Please provide data parameter for this function to be able '
82 'to determine polymorphic scenarios.'.format(table.name)
83 )
84 else:
85 for cls in found_classes:
86 mapper = sa.inspect(cls)
87 polymorphic_on = mapper.polymorphic_on.name
88 if polymorphic_on in data:
89 if data[polymorphic_on] == mapper.polymorphic_identity:
90 return cls
91 raise ValueError(
92 "Multiple declarative classes found for table '{}'. Given "
93 'data row does not match any polymorphic identity of the '
94 'found classes.'.format(table.name)
95 )
96 elif found_classes:
97 return found_classes.pop()
98 return None
101def get_type(expr):
102 """
103 Return the associated type with given Column, InstrumentedAttribute,
104 ColumnProperty, RelationshipProperty or other similar SQLAlchemy construct.
106 For constructs wrapping columns this is the column type. For relationships
107 this function returns the relationship mapper class.
109 :param expr:
110 SQLAlchemy Column, InstrumentedAttribute, ColumnProperty or other
111 similar SA construct.
113 ::
115 class User(Base):
116 __tablename__ = 'user'
117 id = sa.Column(sa.Integer, primary_key=True)
118 name = sa.Column(sa.String)
121 class Article(Base):
122 __tablename__ = 'article'
123 id = sa.Column(sa.Integer, primary_key=True)
124 author_id = sa.Column(sa.Integer, sa.ForeignKey(User.id))
125 author = sa.orm.relationship(User)
128 get_type(User.__table__.c.name) # sa.String()
129 get_type(User.name) # sa.String()
130 get_type(User.name.property) # sa.String()
132 get_type(Article.author) # User
135 .. versionadded: 0.30.9
136 """
137 if hasattr(expr, 'type'):
138 return expr.type
139 elif isinstance(expr, InstrumentedAttribute):
140 expr = expr.property
142 if isinstance(expr, ColumnProperty):
143 return expr.columns[0].type
144 elif isinstance(expr, RelationshipProperty):
145 return expr.mapper.class_
146 raise TypeError("Couldn't inspect type.")
149def cast_if(expression, type_):
150 """
151 Produce a CAST expression but only if given expression is not of given type
152 already.
154 Assume we have a model with two fields id (Integer) and name (String).
156 ::
158 import sqlalchemy as sa
159 from sqlalchemy_utils import cast_if
162 cast_if(User.id, sa.Integer) # "user".id
163 cast_if(User.name, sa.String) # "user".name
164 cast_if(User.id, sa.String) # CAST("user".id AS TEXT)
167 This function supports scalar values as well.
169 ::
171 cast_if(1, sa.Integer) # 1
172 cast_if('text', sa.String) # 'text'
173 cast_if(1, sa.String) # CAST(1 AS TEXT)
176 :param expression:
177 A SQL expression, such as a ColumnElement expression or a Python string
178 which will be coerced into a bound literal value.
179 :param type_:
180 A TypeEngine class or instance indicating the type to which the CAST
181 should apply.
183 .. versionadded: 0.30.14
184 """
185 try:
186 expr_type = get_type(expression)
187 except TypeError:
188 expr_type = expression
189 check_type = type_().python_type
190 else:
191 check_type = type_
193 return (
194 sa.cast(expression, type_)
195 if not isinstance(expr_type, check_type)
196 else expression
197 )
200def get_column_key(model, column):
201 """
202 Return the key for given column in given model.
204 :param model: SQLAlchemy declarative model object
206 ::
208 class User(Base):
209 __tablename__ = 'user'
210 id = sa.Column(sa.Integer, primary_key=True)
211 name = sa.Column('_name', sa.String)
214 get_column_key(User, User.__table__.c._name) # 'name'
216 .. versionadded: 0.26.5
218 .. versionchanged: 0.27.11
219 Throws UnmappedColumnError instead of ValueError when no property was
220 found for given column. This is consistent with how SQLAlchemy works.
221 """
222 mapper = sa.inspect(model)
223 try:
224 return mapper.get_property_by_column(column).key
225 except sa.orm.exc.UnmappedColumnError:
226 for key, c in mapper.columns.items():
227 if c.name == column.name and c.table is column.table:
228 return key
229 raise sa.orm.exc.UnmappedColumnError(
230 f'No column {column} is configured on mapper {mapper}...'
231 )
234def get_mapper(mixed):
235 """
236 Return related SQLAlchemy Mapper for given SQLAlchemy object.
238 :param mixed: SQLAlchemy Table / Alias / Mapper / declarative model object
240 ::
242 from sqlalchemy_utils import get_mapper
245 get_mapper(User)
247 get_mapper(User())
249 get_mapper(User.__table__)
251 get_mapper(User.__mapper__)
253 get_mapper(sa.orm.aliased(User))
255 get_mapper(sa.orm.aliased(User.__table__))
258 Raises:
259 ValueError: if multiple mappers were found for given argument
261 .. versionadded: 0.26.1
262 """
263 if isinstance(mixed, _MapperEntity):
264 mixed = mixed.expr
265 elif isinstance(mixed, sa.Column):
266 mixed = mixed.table
267 elif isinstance(mixed, _ColumnEntity):
268 mixed = mixed.expr
270 if isinstance(mixed, sa.orm.Mapper):
271 return mixed
272 if isinstance(mixed, sa.orm.util.AliasedClass):
273 return sa.inspect(mixed).mapper
274 if isinstance(mixed, sa.sql.selectable.Alias):
275 mixed = mixed.element
276 if isinstance(mixed, AliasedInsp):
277 return mixed.mapper
278 if isinstance(mixed, sa.orm.attributes.InstrumentedAttribute):
279 mixed = mixed.class_
280 if isinstance(mixed, sa.Table):
281 all_mappers = set()
282 for mapper_registry in mapperlib._all_registries():
283 all_mappers.update(mapper_registry.mappers)
284 mappers = [mapper for mapper in all_mappers if mixed in mapper.tables]
285 if len(mappers) > 1:
286 raise ValueError("Multiple mappers found for table '%s'." % mixed.name)
287 elif not mappers:
288 raise ValueError("Could not get mapper for table '%s'." % mixed.name)
289 else:
290 return mappers[0]
291 if not isclass(mixed):
292 mixed = type(mixed)
293 return sa.inspect(mixed)
296def get_bind(obj):
297 """
298 Return the bind for given SQLAlchemy Engine / Connection / declarative
299 model object.
301 :param obj: SQLAlchemy Engine / Connection / declarative model object
303 ::
305 from sqlalchemy_utils import get_bind
308 get_bind(session) # Connection object
310 get_bind(user)
312 """
313 if hasattr(obj, 'bind'):
314 conn = obj.bind
315 else:
316 try:
317 conn = object_session(obj).bind
318 except UnmappedInstanceError:
319 conn = obj
321 if not hasattr(conn, 'execute'):
322 raise TypeError(
323 'This method accepts only Session, Engine, Connection and '
324 'declarative model objects.'
325 )
326 return conn
329def get_primary_keys(mixed):
330 """
331 Return an OrderedDict of all primary keys for given Table object,
332 declarative class or declarative class instance.
334 :param mixed:
335 SA Table object, SA declarative class or SA declarative class instance
337 ::
339 get_primary_keys(User)
341 get_primary_keys(User())
343 get_primary_keys(User.__table__)
345 get_primary_keys(User.__mapper__)
347 get_primary_keys(sa.orm.aliased(User))
349 get_primary_keys(sa.orm.aliased(User.__table__))
352 .. versionchanged: 0.25.3
353 Made the function return an ordered dictionary instead of generator.
354 This change was made to support primary key aliases.
356 Renamed this function to 'get_primary_keys', formerly 'primary_keys'
358 .. seealso:: :func:`get_columns`
359 """
360 return OrderedDict(
361 (
362 (key, column)
363 for key, column in get_columns(mixed).items()
364 if column.primary_key
365 )
366 )
369def get_tables(mixed):
370 """
371 Return a set of tables associated with given SQLAlchemy object.
373 Let's say we have three classes which use joined table inheritance
374 TextItem, Article and BlogPost. Article and BlogPost inherit TextItem.
376 ::
378 get_tables(Article) # set([Table('article', ...), Table('text_item')])
380 get_tables(Article())
382 get_tables(Article.__mapper__)
385 If the TextItem entity is using with_polymorphic='*' then this function
386 returns all child tables (article and blog_post) as well.
388 ::
391 get_tables(TextItem) # set([Table('text_item', ...)], ...])
394 .. versionadded: 0.26.0
396 :param mixed:
397 SQLAlchemy Mapper, Declarative class, Column, InstrumentedAttribute or
398 a SA Alias object wrapping any of these objects.
399 """
400 if isinstance(mixed, sa.Table):
401 return [mixed]
402 elif isinstance(mixed, sa.Column):
403 return [mixed.table]
404 elif isinstance(mixed, sa.orm.attributes.InstrumentedAttribute):
405 return mixed.parent.tables
406 elif isinstance(mixed, _ColumnEntity):
407 mixed = mixed.expr
409 mapper = get_mapper(mixed)
411 polymorphic_mappers = get_polymorphic_mappers(mapper)
412 if polymorphic_mappers:
413 tables = sum((m.tables for m in polymorphic_mappers), [])
414 else:
415 tables = mapper.tables
416 return tables
419def get_columns(mixed):
420 """
421 Return a collection of all Column objects for given SQLAlchemy
422 object.
424 The type of the collection depends on the type of the object to return the
425 columns from.
427 ::
429 get_columns(User)
431 get_columns(User())
433 get_columns(User.__table__)
435 get_columns(User.__mapper__)
437 get_columns(sa.orm.aliased(User))
439 get_columns(sa.orm.alised(User.__table__))
442 :param mixed:
443 SA Table object, SA Mapper, SA declarative class, SA declarative class
444 instance or an alias of any of these objects
445 """
446 if isinstance(mixed, sa.SelectBase):
447 return mixed.subquery().c
448 if isinstance(mixed, sa.sql.selectable.Selectable):
449 return mixed.c
450 if isinstance(mixed, sa.orm.util.AliasedClass):
451 return sa.inspect(mixed).mapper.columns
452 if isinstance(mixed, sa.orm.Mapper):
453 return mixed.columns
454 if isinstance(mixed, InstrumentedAttribute):
455 return mixed.property.columns
456 if isinstance(mixed, ColumnProperty):
457 return mixed.columns
458 if isinstance(mixed, sa.Column):
459 return [mixed]
460 if not isclass(mixed):
461 mixed = mixed.__class__
462 return sa.inspect(mixed).columns
465def table_name(obj):
466 """
467 Return table name of given target, declarative class or the
468 table name where the declarative attribute is bound to.
469 """
470 class_ = getattr(obj, 'class_', obj)
472 try:
473 return class_.__tablename__
474 except AttributeError:
475 pass
477 try:
478 return class_.__table__.name
479 except AttributeError:
480 pass
483def getattrs(obj, attrs):
484 return map(partial(getattr, obj), attrs)
487def quote(mixed, ident):
488 """
489 Conditionally quote an identifier.
490 ::
493 from sqlalchemy_utils import quote
496 engine = create_engine('sqlite:///:memory:')
498 quote(engine, 'order')
499 # '"order"'
501 quote(engine, 'some_other_identifier')
502 # 'some_other_identifier'
505 :param mixed: SQLAlchemy Session / Connection / Engine / Dialect object.
506 :param ident: identifier to conditionally quote
507 """
508 if isinstance(mixed, Dialect):
509 dialect = mixed
510 else:
511 dialect = get_bind(mixed).dialect
512 return dialect.preparer(dialect).quote(ident)
515def _get_query_compile_state(query):
516 return query._compile_state()
519def get_polymorphic_mappers(mixed):
520 if isinstance(mixed, AliasedInsp):
521 return mixed.with_polymorphic_mappers
522 else:
523 return mixed.polymorphic_map.values()
526def get_descriptor(entity, attr):
527 mapper = sa.inspect(entity)
529 for key, descriptor in get_all_descriptors(mapper).items():
530 if attr == key:
531 prop = descriptor.property if hasattr(descriptor, 'property') else None
532 if isinstance(prop, ColumnProperty):
533 if isinstance(entity, sa.orm.util.AliasedClass):
534 for c in mapper.selectable.c:
535 if c.key == attr:
536 return c
537 else:
538 # If the property belongs to a class that uses
539 # polymorphic inheritance we have to take into account
540 # situations where the attribute exists in child class
541 # but not in parent class.
542 return getattr(prop.parent.class_, attr)
543 else:
544 # Handle synonyms, relationship properties and hybrid
545 # properties
547 if isinstance(entity, sa.orm.util.AliasedClass):
548 return getattr(entity, attr)
549 try:
550 return getattr(mapper.class_, attr)
551 except AttributeError:
552 pass
555def get_all_descriptors(expr):
556 if isinstance(expr, sa.sql.selectable.Selectable):
557 return expr.c
558 insp = sa.inspect(expr)
559 try:
560 polymorphic_mappers = get_polymorphic_mappers(insp)
561 except sa.exc.NoInspectionAvailable:
562 return get_mapper(expr).all_orm_descriptors
563 else:
564 attrs = dict(get_mapper(expr).all_orm_descriptors)
565 for submapper in polymorphic_mappers:
566 for key, descriptor in submapper.all_orm_descriptors.items():
567 if key not in attrs:
568 attrs[key] = descriptor
569 return attrs
572def get_hybrid_properties(model):
573 """
574 Returns a dictionary of hybrid property keys and hybrid properties for
575 given SQLAlchemy declarative model / mapper.
578 Consider the following model
580 ::
583 from sqlalchemy.ext.hybrid import hybrid_property
586 class Category(Base):
587 __tablename__ = 'category'
588 id = sa.Column(sa.Integer, primary_key=True)
589 name = sa.Column(sa.Unicode(255))
591 @hybrid_property
592 def lowercase_name(self):
593 return self.name.lower()
595 @lowercase_name.expression
596 def lowercase_name(cls):
597 return sa.func.lower(cls.name)
600 You can now easily get a list of all hybrid property names
602 ::
605 from sqlalchemy_utils import get_hybrid_properties
608 get_hybrid_properties(Category).keys() # ['lowercase_name']
611 This function also supports aliased classes
613 ::
616 get_hybrid_properties(
617 sa.orm.aliased(Category)
618 ).keys() # ['lowercase_name']
621 .. versionchanged: 0.26.7
622 This function now returns a dictionary instead of generator
624 .. versionchanged: 0.30.15
625 Added support for aliased classes
627 :param model: SQLAlchemy declarative model or mapper
628 """
629 return {
630 key: prop
631 for key, prop in get_mapper(model).all_orm_descriptors.items()
632 if isinstance(prop, hybrid_property)
633 }
636def get_declarative_base(model):
637 """
638 Returns the declarative base for given model class.
640 :param model: SQLAlchemy declarative model
641 """
642 for parent in model.__bases__:
643 try:
644 parent.metadata
645 return get_declarative_base(parent)
646 except AttributeError:
647 pass
648 return model
651def getdotattr(obj_or_class, dot_path, condition=None):
652 """
653 Allow dot-notated strings to be passed to `getattr`.
655 ::
657 getdotattr(SubSection, 'section.document')
659 getdotattr(subsection, 'section.document')
662 :param obj_or_class: Any object or class
663 :param dot_path: Attribute path with dot mark as separator
664 """
665 last = obj_or_class
667 for path in str(dot_path).split('.'):
668 getter = attrgetter(path)
670 if is_sequence(last):
671 tmp = []
672 for element in last:
673 value = getter(element)
674 if is_sequence(value):
675 tmp.extend(value)
676 else:
677 tmp.append(value)
678 last = tmp
679 elif isinstance(last, InstrumentedAttribute):
680 last = getter(last.property.mapper.class_)
681 elif last is None:
682 return None
683 else:
684 last = getter(last)
685 if condition is not None:
686 if is_sequence(last):
687 last = [v for v in last if condition(v)]
688 else:
689 if not condition(last):
690 return None
692 return last
695def is_deleted(obj):
696 return obj in sa.orm.object_session(obj).deleted
699def has_changes(obj, attrs=None, exclude=None):
700 """
701 Simple shortcut function for checking if given attributes of given
702 declarative model object have changed during the session. Without
703 parameters this checks if given object has any modificiations. Additionally
704 exclude parameter can be given to check if given object has any changes
705 in any attributes other than the ones given in exclude.
708 ::
711 from sqlalchemy_utils import has_changes
714 user = User()
716 has_changes(user, 'name') # False
718 user.name = 'someone'
720 has_changes(user, 'name') # True
722 has_changes(user) # True
725 You can check multiple attributes as well.
726 ::
729 has_changes(user, ['age']) # True
731 has_changes(user, ['name', 'age']) # True
734 This function also supports excluding certain attributes.
736 ::
738 has_changes(user, exclude=['name']) # False
740 has_changes(user, exclude=['age']) # True
742 .. versionchanged: 0.26.6
743 Added support for multiple attributes and exclude parameter.
745 :param obj: SQLAlchemy declarative model object
746 :param attrs: Names of the attributes
747 :param exclude: Names of the attributes to exclude
748 """
749 if attrs:
750 if isinstance(attrs, str):
751 return sa.inspect(obj).attrs.get(attrs).history.has_changes()
752 else:
753 return any(has_changes(obj, attr) for attr in attrs)
754 else:
755 if exclude is None:
756 exclude = []
757 return any(
758 attr.history.has_changes()
759 for key, attr in sa.inspect(obj).attrs.items()
760 if key not in exclude
761 )
764def is_loaded(obj, prop):
765 """
766 Return whether or not given property of given object has been loaded.
768 ::
770 class Article(Base):
771 __tablename__ = 'article'
772 id = sa.Column(sa.Integer, primary_key=True)
773 name = sa.Column(sa.String)
774 content = sa.orm.deferred(sa.Column(sa.String))
777 article = session.query(Article).get(5)
779 # name gets loaded since its not a deferred property
780 assert is_loaded(article, 'name')
782 # content has not yet been loaded since its a deferred property
783 assert not is_loaded(article, 'content')
786 .. versionadded: 0.27.8
788 :param obj: SQLAlchemy declarative model object
789 :param prop: Name of the property or InstrumentedAttribute
790 """
791 return prop not in sa.inspect(obj).unloaded
794def identity(obj_or_class):
795 """
796 Return the identity of given sqlalchemy declarative model class or instance
797 as a tuple. This differs from obj._sa_instance_state.identity in a way that
798 it always returns the identity even if object is still in transient state (
799 new object that is not yet persisted into database). Also for classes it
800 returns the identity attributes.
802 ::
804 from sqlalchemy import inspect
805 from sqlalchemy_utils import identity
808 user = User(name='John Matrix')
809 session.add(user)
810 identity(user) # None
811 inspect(user).identity # None
813 session.flush() # User now has id but is still in transient state
815 identity(user) # (1,)
816 inspect(user).identity # None
818 session.commit()
820 identity(user) # (1,)
821 inspect(user).identity # (1, )
824 You can also use identity for classes::
827 identity(User) # (User.id, )
829 .. versionadded: 0.21.0
831 :param obj: SQLAlchemy declarative model object
832 """
833 return tuple(
834 getattr(obj_or_class, column_key)
835 for column_key in get_primary_keys(obj_or_class).keys()
836 )
839def naturally_equivalent(obj, obj2):
840 """
841 Returns whether or not two given SQLAlchemy declarative instances are
842 naturally equivalent (all their non primary key properties are equivalent).
845 ::
847 from sqlalchemy_utils import naturally_equivalent
850 user = User(name='someone')
851 user2 = User(name='someone')
853 user == user2 # False
855 naturally_equivalent(user, user2) # True
858 :param obj: SQLAlchemy declarative model object
859 :param obj2: SQLAlchemy declarative model object to compare with `obj`
860 """
861 for column_key, column in sa.inspect(obj.__class__).columns.items():
862 if column.primary_key:
863 continue
865 if not (getattr(obj, column_key) == getattr(obj2, column_key)):
866 return False
867 return True
870def _get_class_registry(class_):
871 return class_.registry._class_registry