Source code for examples.extending_query.filter_public
"""Illustrates a global criteria applied to entities of a particular type.The example here is the "public" flag, a simple boolean that indicatesthe rows are part of a publicly viewable subcategory. Rows that do notinclude this flag are not shown unless a special option is passed to thequery.Uses for this kind of recipe include tables that have "soft deleted" rowsmarked as "deleted" that should be skipped, rows that have access control rulesthat should be applied on a per-request basis, etc."""fromsqlalchemyimportBooleanfromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimporteventfromsqlalchemyimportForeignKeyfromsqlalchemyimportIntegerfromsqlalchemyimportormfromsqlalchemyimportselectfromsqlalchemyimportStringfromsqlalchemyimporttruefromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportrelationshipfromsqlalchemy.ormimportSessionfromsqlalchemy.ormimportsessionmaker@event.listens_for(Session,"do_orm_execute")def_add_filtering_criteria(execute_state):"""Intercept all ORM queries. Add a with_loader_criteria option to all of them. This option applies to SELECT queries and adds a global WHERE criteria (or as appropriate ON CLAUSE criteria for join targets) to all objects of a certain class or superclass. """# the with_loader_criteria automatically applies itself to# relationship loads as well including lazy loads. So if this is# a relationship load, assume the option was set up from the top level# query.if(notexecute_state.is_column_loadandnotexecute_state.is_relationship_loadandnotexecute_state.execution_options.get("include_private",False)):execute_state.statement=execute_state.statement.options(orm.with_loader_criteria(HasPrivate,lambdacls:cls.public==true(),include_aliases=True,))classHasPrivate:"""Mixin that identifies a class as having private entities"""public=Column(Boolean,nullable=False)if__name__=="__main__":Base=declarative_base()classUser(HasPrivate,Base):__tablename__="user"id=Column(Integer,primary_key=True)name=Column(String)addresses=relationship("Address",back_populates="user")classAddress(HasPrivate,Base):__tablename__="address"id=Column(Integer,primary_key=True)email=Column(String)user_id=Column(Integer,ForeignKey("user.id"))user=relationship("User",back_populates="addresses")engine=create_engine("sqlite://",echo=True)Base.metadata.create_all(engine)Session=sessionmaker(bind=engine)sess=Session()sess.add_all([User(name="u1",public=True,addresses=[Address(email="u1a1",public=True),Address(email="u1a2",public=True),],),User(name="u2",public=True,addresses=[Address(email="u2a1",public=False),Address(email="u2a2",public=True),],),User(name="u3",public=False,addresses=[Address(email="u3a1",public=False),Address(email="u3a2",public=False),],),User(name="u4",public=False,addresses=[Address(email="u4a1",public=False),Address(email="u4a2",public=True),],),User(name="u5",public=True,addresses=[Address(email="u5a1",public=True),Address(email="u5a2",public=False),],),])sess.commit()# now querying Address or User objects only gives us the public onesforu1insess.query(User).options(orm.selectinload(User.addresses)):assertu1.public# the addresses collection will also be "public only", which works# for all relationship loaders including joinedloadforaddressinu1.addresses:assertaddress.public# works for columns toocols=(sess.query(User.id,Address.id).join(User.addresses).order_by(User.id,Address.id).all())assertcols==[(1,1),(1,2),(2,4),(5,9)]cols=(sess.query(User.id,Address.id).join(User.addresses).order_by(User.id,Address.id).execution_options(include_private=True).all())assertcols==[(1,1),(1,2),(2,3),(2,4),(3,5),(3,6),(4,7),(4,8),(5,9),(5,10),]# count all public addressesassertsess.query(Address).count()==5# count all addresses public and privateassert(sess.query(Address).execution_options(include_private=True).count()==10)# load an Address that is public, but its parent User is private# (2.0 style query)a1=sess.execute(select(Address).filter_by(email="u4a2")).scalar()# assuming the User isn't already in the Session, it returns Noneasserta1.userisNone# however, if that user is present in the session, then a many-to-one# does a simple get() and it will be presentsess.expire(a1,["user"])u1=sess.execute(select(User).filter_by(name="u4").execution_options(include_private=True)).scalar()asserta1.userisu1