1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""Utils to provide dashboards for tests"""
from typing import Optional
from pandas import DataFrame # noqa: F401
from superset import db
from superset.connectors.sqla.models import SqlaTable
from superset.models.core import Database
from superset.models.dashboard import Dashboard
from superset.models.slice import Slice
from superset.utils import json
from superset.utils.core import DatasourceType, get_example_default_schema
def get_table(
table_name: str,
database: Database,
schema: Optional[str] = None,
):
schema = schema or get_example_default_schema()
return (
db.session.query(SqlaTable)
.filter_by(database_id=database.id, schema=schema, table_name=table_name)
.one_or_none()
)
def create_table_metadata(
table_name: str,
database: Database,
table_description: str = "",
fetch_values_predicate: Optional[str] = None,
schema: Optional[str] = None,
) -> SqlaTable:
schema = schema or get_example_default_schema()
table = get_table(table_name, database, schema)
if not table:
table = SqlaTable(
schema=schema,
table_name=table_name,
normalize_columns=False,
always_filter_main_dttm=False,
)
db.session.add(table)
if fetch_values_predicate:
table.fetch_values_predicate = fetch_values_predicate
table.database = database
table.description = table_description
db.session.commit()
return table
def create_slice(
title: str, viz_type: str, table: SqlaTable, slices_dict: dict[str, str]
) -> Slice:
return Slice(
slice_name=title,
viz_type=viz_type,
datasource_type=DatasourceType.TABLE,
datasource_id=table.id,
params=json.dumps(slices_dict, indent=4, sort_keys=True),
)
def create_dashboard(
slug: str, title: str, position: str, slices: list[Slice]
) -> Dashboard:
dash = db.session.query(Dashboard).filter_by(slug=slug).one_or_none()
if dash:
return dash
dash = Dashboard()
dash.dashboard_title = title
if position is not None:
js = position
pos = json.loads(js)
dash.position_json = json.dumps(pos, indent=4)
dash.slug = slug
if slices is not None:
dash.slices = slices
db.session.add(dash)
db.session.commit()
return dash