(self, owner, scope, kind, has_filter_names)
| 3004 | |
| 3005 | @lru_cache() |
| 3006 | def _comment_query(self, owner, scope, kind, has_filter_names): |
| 3007 | # NOTE: all_tab_comments / all_mview_comments have a row for all |
| 3008 | # object even if they don't have comments |
| 3009 | queries = [] |
| 3010 | if ObjectKind.TABLE in kind or ObjectKind.VIEW in kind: |
| 3011 | # all_tab_comments returns also plain views |
| 3012 | tbl_view = select( |
| 3013 | dictionary.all_tab_comments.c.table_name, |
| 3014 | dictionary.all_tab_comments.c.comments, |
| 3015 | ).where( |
| 3016 | dictionary.all_tab_comments.c.owner == owner, |
| 3017 | dictionary.all_tab_comments.c.table_name.not_like("BIN$%"), |
| 3018 | ) |
| 3019 | if ObjectKind.VIEW not in kind: |
| 3020 | tbl_view = tbl_view.where( |
| 3021 | dictionary.all_tab_comments.c.table_type == "TABLE" |
| 3022 | ) |
| 3023 | elif ObjectKind.TABLE not in kind: |
| 3024 | tbl_view = tbl_view.where( |
| 3025 | dictionary.all_tab_comments.c.table_type == "VIEW" |
| 3026 | ) |
| 3027 | queries.append(tbl_view) |
| 3028 | if ObjectKind.MATERIALIZED_VIEW in kind: |
| 3029 | mat_view = select( |
| 3030 | dictionary.all_mview_comments.c.mview_name.label("table_name"), |
| 3031 | dictionary.all_mview_comments.c.comments, |
| 3032 | ).where( |
| 3033 | dictionary.all_mview_comments.c.owner == owner, |
| 3034 | dictionary.all_mview_comments.c.mview_name.not_like("BIN$%"), |
| 3035 | ) |
| 3036 | queries.append(mat_view) |
| 3037 | if len(queries) == 1: |
| 3038 | query = queries[0] |
| 3039 | else: |
| 3040 | union = sql.union_all(*queries).subquery("tables_and_views") |
| 3041 | query = select(union.c.table_name, union.c.comments) |
| 3042 | |
| 3043 | name_col = query.selected_columns.table_name |
| 3044 | |
| 3045 | if scope in (ObjectScope.DEFAULT, ObjectScope.TEMPORARY): |
| 3046 | temp = "Y" if scope is ObjectScope.TEMPORARY else "N" |
| 3047 | # need distinct since materialized view are listed also |
| 3048 | # as tables in all_objects |
| 3049 | query = query.distinct().join( |
| 3050 | dictionary.all_objects, |
| 3051 | and_( |
| 3052 | dictionary.all_objects.c.owner == owner, |
| 3053 | dictionary.all_objects.c.object_name == name_col, |
| 3054 | dictionary.all_objects.c.temporary == temp, |
| 3055 | ), |
| 3056 | ) |
| 3057 | if has_filter_names: |
| 3058 | query = query.where(name_col.in_(bindparam("filter_names"))) |
| 3059 | return query |
| 3060 | |
| 3061 | @_handle_synonyms_decorator |
| 3062 | def get_multi_table_comment( |
no test coverage detected