tessl/pypi-apache-superset
A modern, enterprise-ready business intelligence web application for data exploration and visualization.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
models-database.mddocs/
Models and Database
SQLAlchemy-based data models and database interface for managing dashboards, charts, datasets, users, and application metadata.
Capabilities
Core Models
Essential data models for Superset's primary entities.
class Dashboard(Model, AuditMixinNullable, ImportExportMixin):
"""Dashboard model for organizing charts and layouts."""
__tablename__ = 'dashboards'
id: int
dashboard_title: str
position_json: str # JSON-encoded dashboard layout
description: str
css: str # Custom CSS styling
slug: str # URL-friendly identifier
published: bool
# Relationships
slices: List[Slice] # Charts in this dashboard
owners: List[User]
roles: List[Role]
def __init__(self, dashboard_title: str = None,
position_json: str = None, **kwargs):
"""
Initialize dashboard.
Args:
dashboard_title: Display name for dashboard
position_json: JSON layout configuration
"""
@property
def table_names(self) -> Set[str]:
"""Get all table names used by charts in this dashboard."""
@property
def datasources(self) -> Set[BaseDatasource]:
"""Get all datasources used by charts in this dashboard."""
def clone(self) -> 'Dashboard':
"""Create a copy of this dashboard."""
class Slice(Model, AuditMixinNullable, ImportExportMixin):
"""Chart/visualization model."""
__tablename__ = 'slices'
id: int
slice_name: str # Chart display name
datasource_id: int
datasource_type: str # 'table' or 'druid'
datasource_name: str
viz_type: str # Chart type (table, bar, line, etc.)
params: str # JSON-encoded chart configuration
description: str
cache_timeout: int
# Relationships
datasource: BaseDatasource
owners: List[User]
dashboards: List[Dashboard]
def __init__(self, slice_name: str = None, viz_type: str = None,
datasource_id: int = None, **kwargs):
"""
Initialize chart.
Args:
slice_name: Display name for chart
viz_type: Visualization type
datasource_id: ID of source dataset
"""
@property
def form_data(self) -> Dict[str, Any]:
"""Get chart configuration as dictionary."""
def get_query_context(self) -> QueryContext:
"""Build query context for data retrieval."""
class Database(Model, AuditMixinNullable, ImportExportMixin):
"""Database connection model."""
__tablename__ = 'dbs'
id: int
database_name: str
sqlalchemy_uri: str # Connection string
password: str # Encrypted password
cache_timeout: int
extra: str # JSON extra configuration
expose_in_sqllab: bool
allow_run_async: bool
allow_csv_upload: bool
allow_ctas: bool
allow_cvas: bool
allow_dml: bool
def __init__(self, database_name: str = None,
sqlalchemy_uri: str = None, **kwargs):
"""
Initialize database connection.
Args:
database_name: Display name for connection
sqlalchemy_uri: SQLAlchemy connection URI
"""
@property
def db_engine_spec(self) -> Type[BaseEngineSpec]:
"""Get database engine specification."""
def get_sqla_engine(self, schema: str = None) -> Engine:
"""Get SQLAlchemy engine for this database."""
def test_connection(self) -> Tuple[bool, str]:
"""Test database connectivity."""Usage Examples:
# Create dashboard
dashboard = Dashboard(
dashboard_title="Sales Dashboard",
position_json='{"CHART-1": {"x": 0, "y": 0, "w": 6, "h": 4}}'
)
db.session.add(dashboard)
db.session.commit()
# Create chart
chart = Slice(
slice_name="Sales by Region",
viz_type="bar",
datasource_id=1,
params='{"metrics": ["sales"], "groupby": ["region"]}'
)
db.session.add(chart)
db.session.commit()Dataset Models
Models for managing data sources and table metadata.
class SqlaTable(Model, BaseDatasource):
"""SQL table dataset model."""
__tablename__ = 'tables'
id: int
table_name: str
main_dttm_col: str # Primary datetime column
default_endpoint: str
database_id: int
fetch_values_predicate: str
is_sqllab_view: bool
template_params: str # JSON template parameters
schema: str
sql: str # Custom SQL for virtual datasets
# Relationships
database: Database
columns: List[TableColumn]
metrics: List[SqlMetric]
owners: List[User]
def __init__(self, table_name: str = None, database: Database = None,
schema: str = None, **kwargs):
"""
Initialize SQL table dataset.
Args:
table_name: Name of database table
database: Database connection
schema: Schema name (optional)
"""
@property
def datasource_name(self) -> str:
"""Get display name for datasource."""
def get_query_str(self, query_obj: QueryObject) -> str:
"""Generate SQL query from query object."""
def get_df(self, query_obj: QueryObject) -> pd.DataFrame:
"""Execute query and return DataFrame."""
class TableColumn(Model, BaseColumn):
"""Table column metadata model."""
__tablename__ = 'table_columns'
id: int
column_name: str
type: str # Data type
groupby: bool # Can be used for grouping
filterable: bool # Can be used for filtering
expression: str # SQL expression for computed columns
description: str
table_id: int
# Relationships
table: SqlaTable
def __init__(self, column_name: str = None, type: str = None,
table: SqlaTable = None, **kwargs):
"""
Initialize table column.
Args:
column_name: Database column name
type: Data type
table: Parent table
"""
class SqlMetric(Model, BaseMetric):
"""SQL metric definition model."""
__tablename__ = 'sql_metrics'
id: int
metric_name: str
expression: str # SQL expression
metric_type: str
description: str
table_id: int
d3format: str # D3 number formatting
# Relationships
table: SqlaTable
def __init__(self, metric_name: str = None, expression: str = None,
table: SqlaTable = None, **kwargs):
"""
Initialize SQL metric.
Args:
metric_name: Display name for metric
expression: SQL expression to calculate metric
table: Parent table
"""User and Security Models
Models for user management, roles, and permissions.
class User(Model):
"""User model for authentication and authorization."""
__tablename__ = 'ab_user'
id: int
first_name: str
last_name: str
username: str
password: str # Hashed password
active: bool
email: str
last_login: datetime
login_count: int
fail_login_count: int
# Relationships
roles: List[Role]
def __init__(self, first_name: str = None, last_name: str = None,
username: str = None, email: str = None, **kwargs):
"""
Initialize user.
Args:
first_name: User's first name
last_name: User's last name
username: Login username
email: Email address
"""
@property
def full_name(self) -> str:
"""Get user's full name."""
class Role(Model):
"""Role model for grouping permissions."""
__tablename__ = 'ab_role'
id: int
name: str
# Relationships
permissions: List[Permission]
users: List[User]
def __init__(self, name: str = None, **kwargs):
"""
Initialize role.
Args:
name: Role name
"""
class Permission(Model):
"""Permission model for access control."""
__tablename__ = 'ab_permission'
id: int
name: str
# Relationships
view_menu: ViewMenu
roles: List[Role]
def __init__(self, name: str = None, **kwargs):
"""
Initialize permission.
Args:
name: Permission name
"""Query and Execution Models
Models for SQL Lab and query execution tracking.
class Query(Model):
"""SQL query execution model."""
__tablename__ = 'query'
id: int
client_id: str
database_id: int
tmp_table_name: str
tab_name: str
sql_editor_id: str
user_id: int
status: str # running, success, failed, etc.
schema: str
sql: str # SQL query text
select_sql: str # Processed SELECT statement
executed_sql: str # Actually executed SQL
limit: int
select_as_cta: bool
select_as_cta_used: bool
progress: int # Query progress percentage
rows: int # Number of rows returned
error_message: str
start_time: datetime
end_time: datetime
# Relationships
database: Database
user: User
def __init__(self, database_id: int = None, sql: str = None,
user_id: int = None, **kwargs):
"""
Initialize query.
Args:
database_id: Target database ID
sql: SQL query to execute
user_id: Executing user ID
"""
@property
def duration(self) -> timedelta:
"""Get query execution duration."""
class SavedQuery(Model, AuditMixinNullable):
"""Saved SQL query model."""
__tablename__ = 'saved_query'
id: int
user_id: int
db_id: int
label: str # Query name
schema: str
sql: str
description: str
# Relationships
database: Database
user: User
def __init__(self, user_id: int = None, db_id: int = None,
label: str = None, sql: str = None, **kwargs):
"""
Initialize saved query.
Args:
user_id: Query owner ID
db_id: Target database ID
label: Display name
sql: SQL query text
"""Utility and Metadata Models
Supporting models for application metadata and configuration.
class KeyValue(Model):
"""Key-value store for application metadata."""
__tablename__ = 'keyvalue'
id: int
value: str # JSON-encoded value
def __init__(self, value: str = None, **kwargs):
"""
Initialize key-value pair.
Args:
value: JSON-encoded value
"""
class CssTemplate(Model, AuditMixinNullable):
"""CSS template for dashboard styling."""
__tablename__ = 'css_templates'
id: int
template_name: str
css: str # CSS content
def __init__(self, template_name: str = None, css: str = None, **kwargs):
"""
Initialize CSS template.
Args:
template_name: Template display name
css: CSS content
"""
class Log(Model):
"""Activity logging model."""
__tablename__ = 'logs'
id: int
action: str
user_id: int
json: str # JSON-encoded log data
dttm: datetime
dashboard_id: int
slice_id: int
# Relationships
user: User
dashboard: Dashboard
slice: Slice
def __init__(self, action: str = None, user_id: int = None,
json: str = None, **kwargs):
"""
Initialize log entry.
Args:
action: Action type
user_id: User performing action
json: Additional log data
"""Database Operations
Session Management
from superset import db
# Database session operations
def create_record(model_instance: Model) -> None:
"""Add new record to database."""
db.session.add(model_instance)
db.session.commit()
def update_record(model_instance: Model) -> None:
"""Update existing record."""
db.session.merge(model_instance)
db.session.commit()
def delete_record(model_instance: Model) -> None:
"""Delete record from database."""
db.session.delete(model_instance)
db.session.commit()
# Query operations
def get_by_id(model_class: Type[Model], id: int) -> Model:
"""Get record by primary key."""
return db.session.query(model_class).get(id)
def get_all(model_class: Type[Model]) -> List[Model]:
"""Get all records of a model type."""
return db.session.query(model_class).all()Model Mixins
class AuditMixinNullable:
"""Mixin for audit fields that can be null."""
created_on: datetime
changed_on: datetime
created_by_fk: int
changed_by_fk: int
# Relationships
created_by: User
changed_by: User
class ImportExportMixin:
"""Mixin for import/export functionality."""
export_parent: str = None
export_children: List[str] = []
def export_schema(self) -> Dict[str, Any]:
"""Export model as dictionary."""
@classmethod
def import_from_dict(cls, session: Session, dict_rep: Dict[str, Any],
recursive: bool = True) -> 'ImportExportMixin':
"""Import model from dictionary."""Install with Tessl CLI
npx tessl i tessl/pypi-apache-superset