Equivalent to .Manager for SqlAlchemy Models

I use SQLAlchemy, and what I liked with Django ORM was a manager that I could implement to override the initial request of the object.

Does something like this exist in SQLAlchemy? I would like to always exclude elements that have "visible = False" when I do something like:

session.query(BlogPost).all() 

Is it possible?

Thanks!

+6
source share
2 answers

EDIT: The original version almost worked. Actually the next version is working.

It looks like you are trying to make the query object be something other than a SELECT table.* FROM table . In sqlalchemy, you can match any "selectable" for a class; However, there are some reservations; if the selectable is not a table, inserting data can be difficult. Something like this is coming up with a workable solution. You probably want to have a standard table displayed to allow inserts, so the first part is a completely normal table, class, and cartographer.

 blog_post_table = Table("blog_posts", metadata, Column('id', Integer, primary_key=True), Column('visible', Boolean, default=True), ... ) class BlogPost(object): pass blog_post_mapper = mapper(BlogPost, blog_post_table) 

Or, if you use a declarative extension, everything will be one

 class BlogPost(Base): __tablename__ = 'blog_posts' id = Column(Integer, primary_key=True) visible = Column(Boolean, default=True) 

Now we need a select expression to represent visible messages.

 visible_blog_posts_expr = sqlalchemy.sql.select( [BlogPost.id, BlogPost.visible]) \ .where(BlogPost.visible == True) \ .alias() 

Or, since the names of all the columns of the requested query are tedious (not to mention the DRY violation), you can use the same construction as session.query(BlogPost) and extract the β€œstatement”. You really do not want this to be associated with the session, so call the class directly.

 visible_blog_posts_expr = \ sqlalchemy.orm.Query(BlogPost) \ .filter(BlogPost.visible == True) \ .statement \ .alias() 

And we will also denote this.

 visible_blog_posts = mapper(BlogPost, visible_blog_posts_expr, non_primary=True) 

You can then use mapper visible_blog_posts instead of BlogPost with Session.query , and you will still get BlogPost , which can be updated and saved as usual.

 posts = session.query(visible_blog_posts).all() assert all(post.visible for post in posts) 

In this particular example, there is not much difference between the explicit use of mapper and the declarative extension, you should still call mapper for non-primary comparisons. In the best case, it allows you to enter SomeClass.colname instead of some_table.c.colname (or SomeClass.__table__.colname , or BlogPost.metadata.tables[BlogPost.__tablename__] or ... etc.).

The errors that I made in the original example are now fixed. I missed the missing [] in the sqlalchemy.sql.select call, which expects the columns to be in sequence. when using the select statement in mapper , sqlalchemy insists that the statement be an alias, so you can call it (SELECT .... ) AS some_subselect_alias_5

+6
source

You can do, for example,

 session.query(BlogPost).filter_by(visible=True) 

which should only provide you with the messages you need.

0
source

Source: https://habr.com/ru/post/892203/


All Articles