D6i^xddlmZddlmZddlmZddlmZddlZ ddl m Z ddl m Z ddlmZmZmZdd lmZdd lmZ dd lmZmZdd lmZdd lmZddlm Z d)dZ!dZ"dZ#dZ$dZ%dZ&dZ'dZ(dZ)dZ*dZ+dZ,dZ-dZ.dZ/dZ0d Z1d!Z2d)d"Z3d#Z4d*d$Z5d%Z6d&Z7d'Z8d(Z9y#e$r dd lmZmZYpwxYw)+) OrderedDict)partial)isclass) attrgetterN)Dialect)hybrid_property)ColumnProperty mapperlibRelationshipProperty)InstrumentedAttribute)UnmappedInstanceError) _ColumnEntity _MapperEntity)object_session) AliasedInsp) is_sequencect|jDchc]}t|dr|j|ur| }}t |dkDr|s$t dj |j|D]H}tj|}|jj}||vs3|||jk(sF|cSt dj |j|r|jSycc}w)a Return declarative class associated with given table. If no class is found this function returns `None`. If multiple classes were found (polymorphic cases) additional `data` parameter can be given to hint which class to return. :: class User(Base): __tablename__ = 'entity' id = sa.Column(sa.Integer, primary_key=True) name = sa.Column(sa.String) get_class_by_table(Base, User.__table__) # User class This function also supports models using single table inheritance. Additional data paratemer should be provided in these case. :: class Entity(Base): __tablename__ = 'entity' id = sa.Column(sa.Integer, primary_key=True) name = sa.Column(sa.String) type = sa.Column(sa.String) __mapper_args__ = { 'polymorphic_on': type, 'polymorphic_identity': 'entity' } class User(Entity): __mapper_args__ = { 'polymorphic_identity': 'user' } # Entity class get_class_by_table(Base, Entity.__table__, {'type': 'entity'}) # User class get_class_by_table(Base, Entity.__table__, {'type': 'user'}) :param base: Declarative model base :param table: SQLAlchemy Table object :param data: Data row to determine the class in polymorphic scenarios :return: Declarative class or None. __table__zMultiple declarative classes found for table '{}'. Please provide data parameter for this function to be able to determine polymorphic scenarios.zMultiple declarative classes found for table '{}'. Given data row does not match any polymorphic identity of the found classes.N) _get_class_registryvalueshasattrrlen ValueErrorformatnamesainspectpolymorphic_onpolymorphic_identitypop)basetabledatac found_classesclsmapperr s a/home/azureuser/techstart-app/venv/lib/python3.12/site-packages/sqlalchemy_utils/functions/orm.pyget_class_by_tabler+sj%T*113 1k "q{{e'; M  =A66 <""?#:#:; <$44K(3Nfu 7M6NN wU  ::e Os !I+5I+ct|dr |j}n t|j}t|ds t d|S#t$r|}Y&wxYw)a. Return the bind for given SQLAlchemy Engine / Connection / declarative model object. :param obj: SQLAlchemy Engine / Connection / declarative model object :: from sqlalchemy_utils import get_bind get_bind(session) # Connection object get_bind(user) bindexecutezSThis method accepts only Session, Engine, Connection and declarative model objects.)rr^rr r2)objconns r*get_bindrb.sg"sFxx !#&++D 4 # )   K% D sA AAcTtdt|jDS)a  Return an OrderedDict of all primary keys for given Table object, declarative class or declarative class instance. :param mixed: SA Table object, SA declarative class or SA declarative class instance :: get_primary_keys(User) get_primary_keys(User()) get_primary_keys(User.__table__) get_primary_keys(User.__mapper__) get_primary_keys(sa.orm.aliased(User)) get_primary_keys(sa.orm.aliased(User.__table__)) .. versionchanged: 0.25.3 Made the function return an ordered dictionary instead of generator. This change was made to support primary key aliases. Renamed this function to 'get_primary_keys', formerly 'primary_keys' .. seealso:: :func:`get_columns` c3BK|]\}}|jr||fywN) primary_key).0r@rFs r* z#get_primary_keys..os) V!!&M s)r get_columnsrDrYs r*get_primary_keysrkOs,>  *51779  r5ct|tjr|gSt|tjr |jgSt|tj j jr|jjSt|tr |j}t|}t|}|rtd|Dg}|S|j}|S)a Return a set of tables associated with given SQLAlchemy object. Let's say we have three classes which use joined table inheritance TextItem, Article and BlogPost. Article and BlogPost inherit TextItem. :: get_tables(Article) # set([Table('article', ...), Table('text_item')]) get_tables(Article()) get_tables(Article.__mapper__) If the TextItem entity is using with_polymorphic='*' then this function returns all child tables (article and blog_post) as well. :: get_tables(TextItem) # set([Table('text_item', ...)], ...]) .. versionadded: 0.26.0 :param mixed: SQLAlchemy Mapper, Declarative class, Column, InstrumentedAttribute or a SA Alias object wrapping any of these objects. c34K|]}|jywre)rX)rgms r*rhzget_tables..s<1ahh%"w E299 % } E266,,BB C||""" E= )   F1&9<(; ::e  $ $$ 77N s EE0/E0ct|d|} |jS#t$rYnwxYw |jjS#t$rYywxYw)z Return table name of given target, declarative class or the table name where the declarative attribute is bound to. r1N)getattr __tablename__rwrr)r`r1s r* table_namer|s^ S(C (F ###     $$$    s  ''A A  A c6ttt||Sre)maprrz)r`attrss r*getattrsrs ww$e ,,r5ct|tr|}nt|j}|j |j |S)a Conditionally quote an identifier. :: from sqlalchemy_utils import quote engine = create_engine('sqlite:///:memory:') quote(engine, 'order') # '"order"' quote(engine, 'some_other_identifier') # 'some_other_identifier' :param mixed: SQLAlchemy Session / Connection / Engine / Dialect object. :param ident: identifier to conditionally quote )r.rrbdialectpreparerquote)rYidentrs r*rrs>*%!5/))   G $ * *5 11r5c>t|dr|jS|S)N_compile_state)rr)querys r*_get_query_compile_stater s!u&'##%% r5cnt|tr |jS|jj Sre)r.rwith_polymorphic_mapperspolymorphic_maprrjs r*rprps-%%---$$++--r5ctj|}t|jD] \}}||k(s t |dr |j nd}t |trt |tjjjr1|jjD]}|j|k(s|ccSt|jj |cSt |tjjjrt||cS t|j |cSy#t"$rYwxYw)Nr/)rrget_all_descriptorsrDrr/r.r rArLrMrOr&r@rzror1rw)entityattrr)r@ descriptorpropr&s r*get_descriptorrs ZZ F.v6<<>Z 3;*1*j*I:&&tD$/fbffkk&>&>?#..00%55D=#$H%#4;;#5#5t<< fbffkk&>&>?"6400"6==$77+,&s$D>> E  E ct|tjjjr |j Stj |} t|}tt|j}|D].}|jjD]\}}||vs |||<0|S#tjj$rt|jcYSwxYwre)r.rrNrOrur&rrpdictr\all_orm_descriptorsrDrBNoInspectionAvailable)r3insprrr submapperr@rs r*rr5s$))445vv ::d D 5d;Z%99:, ,I#,#@#@#F#F#H ,Ze#!+E#J , ,  66 ' '4$3334s B//2C$#C$ct|jjDcic]\}}t|tr||c}}Scc}}w)a Returns a dictionary of hybrid property keys and hybrid properties for given SQLAlchemy declarative model / mapper. Consider the following model :: from sqlalchemy.ext.hybrid import hybrid_property class Category(Base): __tablename__ = 'category' id = sa.Column(sa.Integer, primary_key=True) name = sa.Column(sa.Unicode(255)) @hybrid_property def lowercase_name(self): return self.name.lower() @lowercase_name.expression def lowercase_name(cls): return sa.func.lower(cls.name) You can now easily get a list of all hybrid property names :: from sqlalchemy_utils import get_hybrid_properties get_hybrid_properties(Category).keys() # ['lowercase_name'] This function also supports aliased classes :: get_hybrid_properties( sa.orm.aliased(Category) ).keys() # ['lowercase_name'] .. versionchanged: 0.26.7 This function now returns a dictionary instead of generator .. versionchanged: 0.30.15 Added support for aliased classes :param model: SQLAlchemy declarative model or mapper )r\rrDr.r)rEr@rs r*get_hybrid_propertiesrFsKv$E*>>DDF  C dO , T   sAcx|jD]} |jt|cS|S#t$rY+wxYw)zm Returns the declarative base for given model class. :param model: SQLAlchemy declarative model ) __bases__metadataget_declarative_baserw)rEros r*rrsI //  OO'/ / L   s - 99c|}t|jdD]}t|}t|rBg}|D]8}||}t|r|j |(|j |:|}nCt |tr'||jjj}n |y||}|t|r|D cgc] } || s | }} ||ry|Scc} w)a Allow dot-notated strings to be passed to `getattr`. :: getdotattr(SubSection, 'section.document') getdotattr(subsection, 'section.document') :param obj_or_class: Any object or class :param dot_path: Attribute path with dot mark as separator .N) strsplitrrextendappendr.r r/r)r1) obj_or_classdot_path conditionlastpathgettertmprQvaluevs r* getdotattrrs DH ##C( D! t C &wu%JJu%JJu%  & D 3 4$--..556D \$.s@${3-@c3`K|]%\}}|vr|jj'ywre)historyr)rgr@rexcludes r*rhzhas_changes..s2 T'! LL $ $ & s+.) r.rrrrgetrranyrD)r`rrs` `r*rrsd eS !::c?((,,U3;;GGI I@%@@ @ ?G ZZ_2288:   r5cD|tj|jvS)a Return whether or not given property of given object has been loaded. :: class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) name = sa.Column(sa.String) content = sa.orm.deferred(sa.Column(sa.String)) article = session.query(Article).get(5) # name gets loaded since its not a deferred property assert is_loaded(article, 'name') # content has not yet been loaded since its a deferred property assert not is_loaded(article, 'content') .. versionadded: 0.27.8 :param obj: SQLAlchemy declarative model object :param prop: Name of the property or InstrumentedAttribute )rrunloaded)r`rs r* is_loadedrs6 rzz#// //r5cZtfdtjDS)a Return the identity of given sqlalchemy declarative model class or instance as a tuple. This differs from obj._sa_instance_state.identity in a way that it always returns the identity even if object is still in transient state ( new object that is not yet persisted into database). Also for classes it returns the identity attributes. :: from sqlalchemy import inspect from sqlalchemy_utils import identity user = User(name='John Matrix') session.add(user) identity(user) # None inspect(user).identity # None session.flush() # User now has id but is still in transient state identity(user) # (1,) inspect(user).identity # None session.commit() identity(user) # (1,) inspect(user).identity # (1, ) You can also use identity for classes:: identity(User) # (User.id, ) .. versionadded: 0.21.0 :param obj: SQLAlchemy declarative model object c36K|]}t|ywre)rz)rg column_keyrs r*rhzidentity..Ks    j)r)tuplerkkeys)rs`r*identityr$s.N *<8==? r5ctj|jjj D]-\}}|j rt ||t ||k(r-yy)a Returns whether or not two given SQLAlchemy declarative instances are naturally equivalent (all their non primary key properties are equivalent). :: from sqlalchemy_utils import naturally_equivalent user = User(name='someone') user2 = User(name='someone') user == user2 # False naturally_equivalent(user, user2) # True :param obj: SQLAlchemy declarative model object :param obj2: SQLAlchemy declarative model object to compare with `obj` FT)rrrxr0rDrfrz)r`obj2rrFs r*naturally_equivalentrQs^,!jj7??EEG F    Z(GD*,EE  r5cf |jjS#t$r|jcYSwxYwre)registry_class_registryrw_decl_class_registry)r1s r*rrps2+... +***+s 00re)NN): collectionsr functoolsrrroperatorr sqlalchemyrsqlalchemy.engine.interfacesrsqlalchemy.ext.hybridrsqlalchemy.ormr r r sqlalchemy.orm.attributesr sqlalchemy.orm.excr sqlalchemy.orm.contextrr ImportErrorsqlalchemy.orm.querysqlalchemy.orm.sessionrsqlalchemy.orm.utilrutilsrr+r4r=rGr\rbrkrsrir|rrrrprrrrrrrrrrrr5r*rs#01JJ;4BC2+M`-.`0fD>BB%P/d,%^ $-28.:"=@ )X5> B0<*Z>+CBAABsB)) B98B9