使用自定義引數方式實現 superset 實現SQL動態查詢

1、啟用引數：config.py 設定"ENABLE_TEMPLATE_PROCESSING": True

2、當前superset v1.2版本支援的引數包括：

{{ current_username() }} 當前登入使用者名稱
{{ current_username(add_to_cache_keys=False) }} 不從快取中獲取登入使用者名稱，預設從快取獲取
{{ current_user_id()}} 當前登入使用者ID
{{ current_user_id(add_to_cache_keys=False) }}不從快取中獲取登入使用者ID，預設從快取獲取
{{ url_param('custom_variable') }} url 引數，比如127.0.0.1：8001\dashboard?abc=123，引數就是{{ url_param('abc') }} 結果就是123
{{ cache_key_wrapper() }} 還沒有弄明白啥用
{{ filter_values("欄位名") }} 獲取dashboard filter_box元件對某個欄位的篩選結果
{{ from_dttm }} 獲取dashboard filter_box元件日期篩選的開始時間
{{ to_dttm }} 獲取dashboard filter_box元件日期篩選的結束時間
{{ get_filters() }} 暫時沒有弄明白

除此之外，還可以自定義引數，自定義引數方法：

①修改superset/jinja_context.py檔案，修改三個地方：

 1 regex = re.compile(
 2         r"\{\{.*("
 3         r"current_user_id\(.*\)|"
 4         r"current_username\(.*\)|"
 5         r"current_userroles\(.*\)|"
 6         r"isadmin\(.*\)|"
 7         r"cache_key_wrapper\(.*\)|"
 8         r"
 
url_param\(.*\)"
 9         r").*\}\}"
10     )

↑↑↑↑注意此處的current_userroles 和isadmin 是我自定義的，原始檔沒有

 1     def current_user_id(self, add_to_cache_keys: bool = True) -> Optional[int]:
 2         """
 3         Return the user ID of the user who is currently logged in.
 4 
 5         :param add_to_cache_keys: Whether the value should be included in the cache key
 
 6         :returns: The user ID
 7         """
 8 
 9         if hasattr(g, "user") and g.user:
10             if add_to_cache_keys:
11                 self.cache_key_wrapper(g.user.get_id())
12             return g.user.get_id()
13         return None
14 
15     def current_username(self, add_to_cache_keys: bool = True) -> Optional[str]:
16         """
17         Return the username of the user who is currently logged in.
18 
19         :param add_to_cache_keys: Whether the value should be included in the cache key
20         :returns: The username
21         """
22 
23         if g.user and hasattr(g.user, "username"):
24             if add_to_cache_keys:
25                 self.cache_key_wrapper(g.user.username)
26             return g.user.username
27         return None
28     def current_userroles(self, add_to_cache_keys: bool = True) -> Optional[str]:
29         """
30         Return the roles of the user who is currently logged in.
31 
32         :param add_to_cache_keys: Whether the value should be included in the cache key
33         :returns: The userroles
34         """
35 
36         if g.user and hasattr(g.user, "roles"):
37             if add_to_cache_keys:
38                 user_roles = "/".join([role.name.lower() for role in list(g.user.roles)])
39                 self.cache_key_wrapper(user_roles)
40                 print(user_roles)
41                 return user_roles
42                 """admin in user_roles"""
43         return None
44 
45     def isadmin(self, add_to_cache_keys: bool = True) -> Optional[str]:
46         """
47         Return the roles of the user who is currently logged in.
48 
49         :param add_to_cache_keys: Whether the value should be included in the cache key
50         :returns: The userroles
51         """
52 
53         if g.user and hasattr(g.user, "roles"):
54             if add_to_cache_keys:
55                 user_roles = [role.name.lower() for role in list(g.user.roles)]
56                 return "admin" in user_roles
57         return None

↑↑↑↑仿照系統自帶的current_username 編造自己的函式，我寫了current_userroles 和isadmin

 1 class JinjaTemplateProcessor(BaseTemplateProcessor):
 2     def set_context(self, **kwargs: Any) -> None:
 3         super().set_context(**kwargs)
 4         extra_cache = ExtraCache(self._extra_cache_keys)
 5         self._context.update(
 6             {
 7                 "url_param": partial(safe_proxy, extra_cache.url_param),
 8                 "current_user_id": partial(safe_proxy, extra_cache.current_user_id),
 9                 "current_username": partial(safe_proxy, extra_cache.current_username),
10                 "current_userroles": partial(safe_proxy, extra_cache.current_userroles),
11                 "isadmin": partial(safe_proxy, extra_cache.isadmin),
12                 "cache_key_wrapper": partial(safe_proxy, extra_cache.cache_key_wrapper),
13                 "filter_values": partial(safe_proxy, filter_values),
14             }
15         )

↑↑↑↑仿照系統自帶的current_username 編造自己的函式，我寫了current_userroles 和isadmin

就是這3個地方，但是注意，自己在第二步早的函式，返回值必須是：

 1 ALLOWED_TYPES = (
 2     NONE_TYPE,
 3     "bool",
 4     "str",
 5     "unicode",
 6     "int",
 7     "long",
 8     "float",
 9     "list",
10     "dict",
11     "tuple",
12     "set",
13 )

否則會提示錯誤，或者自己修改這個types，我是轉換，比如上面那個g.user.roles 返回的結果就不是上面型別，導致我一直不成功，最後修改了下，才可以

3、判斷是否自定義成功：

在superset sql lab中執行如下程式碼，如果能被解析，就說明成功

4、應用案例：

在dataset裡面，動態訪問資料來源，資料來源新增where語句：select * from sales where salesname =' {{current_username()}}'

dashboard裡面，通過獲取篩選器的結果，然後獲取其他表應當顯示的資料範圍：

 1 select  DATE,risktype,sum(num) as num from 
 2 (SELECT date , customerid,product,risktype ,count(*) as num 
 3 from v_superset_forecast_risk group by date , customerid,product,risktype ) a
 4 join 
 5 (select distinct customer_code,product from v_superset_access
 6 where name='{{ current_username() }}' )access
 7 on a.customerid=access.customer_code
 8 and a.product=access.product
 9 and DATE_FORMAT(date,'%Y-%m')> DATE_FORMAT(date_sub(STR_TO_DATE(concat( {{ "'" + "', '".join(filter_values('yearmonthend')) + "'" }},'-01'), '%Y-%m-%d'), interval 12 month),'%Y-%m')
10 and DATE_FORMAT(date,'%Y-%m')<={{ "'" + "', '".join(filter_values('yearmonthend')) + "'" }}
11 group by DATE,risktype

因為sql裡面可以使用jinja 表示式，比如判斷篩選當前沒有篩選的時候，獲取什麼資料

注意{% %} 內部使用引數的時候，不需要加{{}}，否則報錯

通過篩選器實現模糊查詢

5、官方參考文件：https://superset.apache.org/docs/installation/sql-templating

官方沒有那麼詳細，但是裡面有一些我這裡可能也沒有消化吸收掉，可以參考看下

總之，通過上面的自定義引數方法，和jinja表示式在sql中的應用，可以實現動態查詢，解決一些無法通過頁面直接互動查詢結果顯示的內容

另外如果你有其他應用或者自定義上的思考，歡迎留言，相互學習