# Flask實作_ext_18_Flask-SQLAlchemy_Query ###### tags: `flask` `flask_ext` `python` `sqlalchemy` SQLAlchemy的query類有著很多的method,了解它對我們利用orm操作資料庫有著非常大的幫助,這邊也單純的記錄著個人可以理解的method說明。 註:以專案中的資料表為例 註:Blog_Main是blog的主表、Blog_Post是blog的明細表 ## 作業說明 ### add_column ### add_columns ### add_entity ### [all](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.all) * 回傳所有查詢結果 ```python= >>> post = Blog_Post.query.all() >>> len(post) 17 >>> type(post) <class 'list'> ``` ### as_scalar ### autoflush ### [column_descriptions](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.column_descriptions) * 回傳資料表的欄位屬性說明 ```python= >>> post = Blog_Main.query.with_entities(Blog_Main.id, Blog_Main.author) >>> pprint(post.column_descriptions) [{'aliased': False, 'entity': <class 'app_blog.blog.model.Blog_Main'>, 'expr': <sqlalchemy.orm.attributes.InstrumentedAttribute object at 0x05661540>, 'name': 'id', 'type': Integer()}, {'aliased': False, 'entity': <class 'app_blog.blog.model.Blog_Main'>, 'expr': <sqlalchemy.orm.attributes.InstrumentedAttribute object at 0x05661870>, 'name': 'author', 'type': Integer()}] ``` ### correlate ### [count](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.count) * 效果同sql的count ```python= >>> post = Blog_Post.query >>> post.count() 17 ``` ```python= post = Blog_Post.query.filter_by(blog_main_id=1) post.count() 10 ``` ### cte ### delete ### distinct ### enable_assertions ### enable_eagerloads ### except_ ### except_all ### execution_options ### exists ### [filter](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.filter) * 能夠較為直述式的使用語法 * filter_by無法使用>,< * 等號的話要使用== ```python= >>> post = Blog_Post.query.filter(Blog_Post.id>1, Blog_Post.blog_main_id==1) >>> str(post.statement) 'SELECT "Blog_Posts".id, "Blog_Posts".title, "Blog_Posts".body, "Blog_Posts".category_id, "Blog_Posts".author_id, "Blog_Posts".blog_main_id, "Blog_Posts".create_date, "Blog_Posts".edit_date, "Blog_Posts".slug, "Blog_Posts".flag \nFROM "Blog_Posts" WHERE "Blog_Posts".id > :id_1 AND "Blog_Posts".blog_main_id = :blog_main_id_1' ``` ### [filter_by](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.filter_by) * 如同sql的where * 有人說是語法糖,可以用來做簡單的查詢 ```python= >>> post = Blog_Post.query.filter_by(id=1).all() >>> post [<POST> Flask] ``` * 多條件 ```python= >>> post = Blog_Post.query.filter_by(id=1, blog_main_id=1) >>> str(post.statement) 'SELECT "Blog_Posts".id, "Blog_Posts".title, "Blog_Posts".body, "Blog_Posts".category_id, "Blog_Posts".author_id, "Blog_Posts".blog_main_id, "Blog_Posts".create_date, "Blog_Posts".edit_date, "Blog_Posts".slug, "Blog_Posts".flag \nFROM "Blog_Posts" WHERE "Blog_Posts".id = :id_1 AND "Blog_Posts".blog_main_id = :blog_main_id_1' ``` ### [first](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.first) * 回傳第一筆資料,若無查詢資料則回傳None ```python= >>> post = Blog_Post.query.first() >>> post <POST> Flask ``` ### [from_self](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.from_self) * 把查詢結果回傳給自己做子查詢 ```python= >>> post = Blog_Main.query.with_entities(Blog_Main.id, Blog_Main.author) >>> print(post.statement) SELECT "Blog_Main".id, "Blog_Main".author FROM "Blog_Main" >>> print(post.from_self().statement) SELECT anon_1."Blog_Main_id", anon_1."Blog_Main_author" FROM (SELECT "Blog_Main".id AS "Blog_Main_id", "Blog_Main".author AS "Blog_Main_author" FROM "Blog_Main") AS anon_1 ``` ```python= >>> post = Blog_Main.query.filter(Blog_Main.blog_name.like('Flask%')).with_entities(Blog_Main.id, Blog_Main.author) >>> print(post.from_self().statement) SELECT anon_1."Blog_Main_id", anon_1."Blog_Main_author" FROM (SELECT "Blog_Main".id AS "Blog_Main_id", "Blog_Main".author AS "Blog_Main_author" FROM "Blog_Main" WHERE "Blog_Main".blog_name LIKE :blog_name_1) AS anon_1 ``` ### [from_statement](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.from_statement) * 直接使用標準sql語法查詢 * 需配合`from sqlalchemy import text` ```python= >>> from sqlalchemy import text >>> str = 'select id from Blog_posts' >>> post = Blog_Post.query.from_statement(text(str)) >>> print(post.statement) select id from Blog_posts ``` ```python= >>> from sqlalchemy import text >>> str = 'select id from Blog_posts' >>> post = Blog_Main.query.from_statement(text(str)) >>> print(post.statement) select id from Blog_posts ``` 雖然是Blog_Main,但是沒有實際query之前這語法還是會過,後面只要接了all、first...就會異常。 ### [get](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.get) * 取回get內的索引值資料,若無則回傳None ```python= >>> post = Blog_Post.query.get(1) >>> post <POST> Flask >>> post.id 1 ``` ### [group_by](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.group_by) * 效果如標準sql語法的gorup_by ```python= >>> post = Blog_Post.query.with_entities(Blog_Post.blog_main_id).group_by(Blog_Post.blog_main_id) >>> print(post.statement) SELECT "Blog_Posts".blog_main_id FROM "Blog_Posts" GROUP BY "Blog_Posts".blog_main_id ``` ### [having](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.having) * 效果如標準sql語法的having * 需`from sqlalchemy import func` ```python= >>> post = Blog_Post.query.with_entities(Blog_Post.blog_main_id).group_by(Blog_Post.blog_main_id).having(func.count(Blog_Post.blog_main_id)>1) >>> print(post.statement) SELECT "Blog_Posts".blog_main_id FROM "Blog_Posts" GROUP BY "Blog_Posts".blog_main_id HAVING count("Blog_Posts".blog_main_id) > :count_1 ``` ### instances ### intersect ### intersect_all ### [join](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.join) * join_relationship * 利用Model設置的relationship來做join ```python= >>> blog = Blog_Main.query.join(Blog_Main.posts) >>> print(blog.statement) SELECT "Blog_Main".id, "Blog_Main".blog_name, "Blog_Main".blog_descri, "Blog_Main".blog_create_date, "Blog_Main".blog_cover_url, "Blog_Main".author FROM "Blog_Main" JOIN "Blog_Posts" ON "Blog_Main".id = "Blog_Posts".blog_main_id ``` * join_entity * 利用模型來做join,會自動以fk來做join * 如果模型間沒有fk的話會造成異常 ```python= >>> blog = Blog_Main.query.join(Blog_Post) >>> print(blog.statement) SELECT "Blog_Main".id, "Blog_Main".blog_name, "Blog_Main".blog_descri, "Blog_Main".blog_create_date, "Blog_Main".blog_cover_url, "Blog_Main".author FROM "Blog_Main" JOIN "Blog_Posts" ON "Blog_Main".id = "Blog_Posts".blog_main_id ``` 下面是故意以沒有fk值的model來做join ```python= >>> blog = Blog_Main.query.join(Role) Traceback (most recent call last): File "D:\proPycharm\app_blog\venv\lib\site-packages\sqlalchemy\orm\query.py", line 2386, in _join_to_left clause, right, onclause, isouter=outerjoin, full=full) ``` * join_on * 手動設置join on ```python= >>> blog = Blog_Main.query.join(Blog_Post, Blog_Main.id==Blog_Post.blog_main_id) >>> print(blog.statement) SELECT "Blog_Main".id, "Blog_Main".blog_name, "Blog_Main".blog_descri, "Blog_Main".blog_create_date, "Blog_Main".blog_cover_url, "Blog_Main".author FROM "Blog_Main" JOIN "Blog_Posts" ON "Blog_Main".id = "Blog_Posts".blog_main_id ``` * join_subquery * 可將另一個subquery物件拿來做join * 有設置relationship ```python= >>> blog = Blog_Main.query.filter_by(author=1).subquery() >>> user = UserRegister.query.join(blog, UserRegister.blog_mains) >>> print(user.statement) SELECT "UserRgeisters".id, "UserRgeisters".username, "UserRgeisters".email, "UserRgeisters".password_hash, "UserRgeisters".confirm, "UserRgeisters".about_me, "UserRgeisters".location, "UserRgeisters".gender, "UserRgeisters".regist_date, "UserRgeisters".last_login FROM "UserRgeisters" JOIN (SELECT "Blog_Main".id AS id, "Blog_Main".blog_name AS blog_name, "Blog_Main".blog_descri AS blog_descri, "Blog_Main".blog_create_date AS blog_create_date, "Blog_Main".blog_cover_url AS blog_cover_url, "Blog_Main".author AS author FROM "Blog_Main" WHERE "Blog_Main".author = :author_1) AS anon_1 ON "UserRgeisters".id = anon_1.author ``` 或是利用一般的on的作法 ```python= user = UserRegister.query.join(blog, blog.c.author==UserRegister.id) ``` [c的意義](http://docs.sqlalchemy.org/en/latest/core/selectable.html#sqlalchemy.sql.expression.FromClause.c) * join_select_from * 注意到join是以`select_from`為主,不再是`query`的物件 ```python= >>> user = UserRegister.query.select_from(Blog_Main).join(Blog_Main.user) >>> print(user.statement) SELECT "UserRgeisters".id, "UserRgeisters".username, "UserRgeisters".email, "UserRgeisters".password_hash, "UserRgeisters".confirm, "UserRgeisters".about_me, "UserRgeisters".location, "UserRgeisters".gender, "UserRgeisters".regist_date, "UserRgeisters".last_login FROM "Blog_Main" JOIN "UserRgeisters" ON "UserRgeisters".id = "Blog_Main".author ``` ### label ### lazy_loaded_from ### limit ### merge_result ### offset ### [one](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.one) * 回傳一筆資料集,但是查詢結果若超過一筆資料或無資料會拋出異常 * `sqlalchemy.orm.exc.MultipleResultsFound` * `sqlalchemy.orm.exc.NoResultFound` ```python= >>> user = UserRegister.query.filter_by(id=1).one() >>> user username:Shaoe.chen ``` ### [one_or_none](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.one_or_none) * 回傳一筆資料集,但是查詢結果若超過一筆資料會拋出異常,無資料則回傳None ```python= >>> user = UserRegister.query.filter_by(id=11).one_or_none() >>> type(user) <class 'NoneType'> ``` ### [only_return_tuples](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.only_return_tuples) * 回傳tuple,應用於對單一元素的查詢 ```python= >>> user = UserRegister.query.only_return_tuples(True) >>> user <flask_sqlalchemy.BaseQuery object at 0x0060CF30> >>> for u in user: ... print(u) ... (username:Shaoe.chen,) ``` ### options ### [order_by](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.order_by) * 排序 ```python= >>> post = Blog_Post.query.order_by(Blog_Post.blog_main_id) >>> print(post.statement) SELECT "Blog_Posts".id, "Blog_Posts".title, "Blog_Posts".body, "Blog_Posts".category_id, "Blog_Posts".author_id, "Blog_Posts".blog_main_id, "Blog_Posts".create_date, "Blog_Posts".edit_date, "Blog_Posts".slug, "Blog_Posts".flag FROM "Blog_Posts" ORDER BY "Blog_Posts".blog_main_id ``` ### [outerjoin](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.outerjoin) * 用法與join同,就是select的主表不同的差異而以 * 創建一個left outer join的關聯 ### [params](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.params) * 搜尋中以參數來做條件 ```python >>> role = Role.query.filter("id=:id").params(id=1) >>> print(role.statement) SELECT roles.id, roles.name FROM roles WHERE id=:id >>> role.all() [Role is ADMIN] ``` ### populate_existing ### prefix_with ### reset_joinpoint ### scalar ### select_entity_from ### [select_from](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.select_from) * 通常跟join搭配一起使用,可看join的使用說明 ### selectable ### slice ### [statement](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.statement) * 回傳解析的sql語法 ```python= >>> post = Blog_Post.query >>> print(post.statement) SELECT "Blog_Posts".id, "Blog_Posts".title, "Blog_Posts".body, "Blog_Posts".category_id, "Blog_Posts".author_id, "Blog_Posts".blog_main_id, "Blog_Posts".create_date, "Blog_Posts".edit_date, "Blog_Posts".slug, "Blog_Posts".flag FROM "Blog_Posts" ``` ### subquery ### suffix_with ### union ### union_all ### update ### value ### values ### whereclause ### [with_entities](http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.with_entities) * 可將query物件再來做再一次的搜尋,在flask_sqlalchemy中可以用來做搜尋欄位選擇 ```python= >>> post = Blog_Post.query.with_entities(Blog_Post.id, Blog_Post.author_id) >>> print(post.statement) SELECT "Blog_Posts".id, "Blog_Posts".author_id FROM "Blog_Posts" ``` [參考_stackoverflow](https://stackoverflow.com/questions/11530196/flask-sqlalchemy-query-specify-column-names) ### with_for_update ### with_hint ### with_labels ### with_lockmode ### with_parent ### with_polymorphic ### with_session ### with_statement_hint ### with_transformation ### yield_per