Core Module

The core module contains the fundamental classes for building dashboards with Dashboard Lego.

Data Sources

This module defines the base data source with stateless 2-stage pipeline.

hierarchy:

[Core | DataSources | DataSource]

relates-to:

  • motivated_by: “v0.15.0 Refactor: Stateless architecture with 2-stage pipeline + optional lambda functions”

  • implements: “interface: ‘DataSource’ with stateless pipeline and lambda function support”

  • uses: [“library: ‘diskcache’”, “class: ‘DataBuilder’”, “class: ‘DataFilter’”]

contract:

  • pre: “data_builder and data_filter provided OR build_fn/transform_fn for simple cases”

  • post: “get_processed_data(params) runs Build → Filter pipeline”

  • invariant: “NO stored data state, NO abstract methods, only cache”

complexity:

7

decision_cache:

“2-stage pipeline (Build → Filter) for semantic clarity + lambda functions for simplicity”

Example usage with lambda functions:
>>> # Simple datasource with lambda functions
>>> ds = DataSource(
...     build_fn=lambda params: pd.DataFrame({'x': [1, 2, 3]}),
...     transform_fn=lambda df: df * 2
... )
>>> result = ds.get_processed_data()
>>>
>>> # More complex with parameters
>>> ds = DataSource(
...     build_fn=lambda params: pd.read_csv(params.get('file', 'default.csv')),
...     transform_fn=lambda df: df.groupby('category').sum().reset_index()
... )
>>> result = ds.get_processed_data({'file': 'sales.csv'})
>>>
>>> # With cache prewarming
>>> ds = DataSource(
...     build_fn=lambda params: pd.read_csv(params.get('file', 'default.csv')),
...     transform_fn=lambda df: df.groupby('category').sum().reset_index(),
...     cache_prewarm_params=[
...         {'file': 'sales.csv', 'category': 'Electronics'},
...         {'file': 'sales.csv', 'category': 'Home'}
...     ]
... )
>>> # Cache is now prewarmed for these parameter combinations
class dashboard_lego.core.datasource.DataSource(data_builder: Any | None = None, data_transformer: Any | None = None, param_classifier: Callable[[str], str] | None = None, cache_dir: str | None = None, cache_ttl: int = 300, cache_backend: str | Any | None = None, build_fn: Callable[[Dict[str, Any]], DataFrame] | None = None, transform_fn: Callable[[DataFrame], DataFrame] | None = None, cache_prewarm_params: List[Dict[str, Any]] | None = None, df: DataFrame | None = None, **kwargs)[source]

Bases: object

Base data source with stateless 2-stage pipeline (synchronous only).

Pipeline stages (all via cache): 1. Build (DataBuilder.build) - Load + Process 2. Filter (DataFilter.filter) - Apply filters

NO STORED STATE - data computed fresh each call via cache. NO ABSTRACT METHODS - fully concrete base class. SYNC ONLY - for async support, use AsyncDataSource instead.

Hierarchy:

[Core | DataSources | DataSource]

Relates-to:

  • motivated_by: “v0.15.0: 2-stage pipeline for semantic clarity”

  • implements: “class: ‘DataSource’ stateless (sync only)”

  • uses: [“library: ‘diskcache’”, “class: ‘DataBuilder’”, “class: ‘DataFilter’”]

Rationale:

“2-stage pipeline (Build → Filter) simpler than 3-stage”

Contract:

  • pre: “data_builder and data_filter provided (sync only)”

  • post: “get_processed_data(params) returns filtered data via cache (sync)”

  • invariant: “No stored data attributes, no abstract methods, sync only”

Complexity:

7

Decision_cache:

“Chose 2-stage over 3-stage for semantic clarity. Async methods removed - use AsyncDataSource for async support.”

__init__(data_builder: Any | None = None, data_transformer: Any | None = None, param_classifier: Callable[[str], str] | None = None, cache_dir: str | None = None, cache_ttl: int = 300, cache_backend: str | Any | None = None, build_fn: Callable[[Dict[str, Any]], DataFrame] | None = None, transform_fn: Callable[[DataFrame], DataFrame] | None = None, cache_prewarm_params: List[Dict[str, Any]] | None = None, df: DataFrame | None = None, **kwargs)[source]

Initialize datasource with 2-stage pipeline.

Hierarchy:

[Core | DataSources | DataSource | Initialization]

Relates-to:

  • motivated_by: “v0.15.0: 2-stage pipeline configuration with optional lambda functions”

  • implements: “method: ‘__init__’”

Contract:

  • pre: “df, build_fn, and data_builder are optional (if none provided, uses default DataBuilder)”

  • post: “2-stage pipeline ready with handlers created from df/build_fn/builder if provided, or default DataBuilder”

  • stages: “Build → Transform”

  • priority: “df → build_fn → data_builder → default DataBuilder (first provided wins)”

Parameters:

  • data_builder – DataBuilder for stage 1 (load + process). If None and build_fn or df provided, creates appropriate builder.

  • data_transformer – DataTransformer for stage 2 (filtering/aggregation). If None and transform_fn provided, creates LambdaTransformer.

  • param_classifier – Routes params: ‘build’ or ‘transform’. Default: ‘build__’ → ‘build’, ‘transform__’ → ‘transform’.

  • cache_dir – Directory for disk cache. If None, uses in-memory cache. Ignored if cache_backend is provided.

  • cache_ttl – Time-to-live for cache entries in seconds.

  • cache_backend – Cache backend to use. Can be: - ‘disk’ or None: DiskCacheBackend (default, uses cache_dir) - ‘redis’: RedisCacheBackend (localhost:6379) - ‘memory’: InMemoryCacheBackend - CacheBackend instance: Custom backend

  • build_fn – Optional lambda function for simple data building: Dict[str, Any] → DataFrame. If provided, creates LambdaBuilder automatically. Signature: lambda params: df

  • transform_fn – Optional lambda function for simple data transformation: DataFrame → DataFrame. If provided, creates LambdaTransformer automatically. Signature: lambda df: df

  • cache_prewarm_params – Optional list of parameter dictionaries to prewarm cache during initialization. Each dict will be processed through the 2-stage pipeline to populate cache.

  • df – Optional pandas DataFrame. If provided, automatically creates DfHandler as builder. If none of df/build_fn/data_builder provided, uses default DataBuilder (returns empty DataFrame).

with_builder(builder: Any | Callable) → DataSource[source]

Return new datasource with replaced builder.

Immutable pattern for flexible data pipeline composition.

Hierarchy:

[Core | DataSources | DataSource | WithBuilder]

Contract:

  • pre: “builder is DataBuilder instance”

  • post: “Returns new DataSource (does NOT modify self)”

Complexity:

2

Parameters:

builder – DataBuilder instance

Returns:

New DataSource with specified builder

with_builder_fn(build_fn: Callable[[Dict[str, Any]], DataFrame]) → DataSource[source]

Returns a new datasource instance with a lambda-based builder.

Convenience factory for simple build logic without creating DataBuilder class. Symmetric to with_transform_fn() for consistency.

Hierarchy:

[Core | DataSources | DataSource | WithBuilderFn]

Relates-to:

  • motivated_by: “v0.15.0: Symmetric API with with_transform_fn()”

  • implements: “method: ‘with_builder_fn’”

  • uses: [“class: ‘DataBuilder’”]

Rationale:

“Lambda-based builder for simple cases, avoiding class boilerplate”

Contract:

  • pre: “build_fn is callable: Dict[str, Any] → DataFrame”

  • post: “Returns new DataSource with lambda builder”

  • invariant: “Original datasource unchanged (immutable)”

Complexity:

2

Decision_cache:

“Symmetric with_builder_fn/with_transform_fn API for consistency”

Parameters:

build_fn – Function that builds DataFrame from params. Signature: lambda params: df Examples: - lambda p: pd.read_csv(p[‘file_path’]) - lambda p: generate_sample_data(n=p.get(‘rows’, 100))

Returns:

New DataSource instance with lambda builder. Original datasource is unchanged.

Example

>>> # Create datasource with lambda builder
>>> ds = DataSource().with_builder_fn(
...     lambda params: pd.read_csv('data.csv')
... )
>>>
>>> # Or with params
>>> ds = DataSource().with_builder_fn(
...     lambda params: pd.read_csv(params.get('file', 'default.csv'))
... )
with_transformer(transformer: Any) → DataSource[source]

Return new datasource with replaced transformer.

Immutable pattern for flexible data pipeline composition.

Hierarchy:

[Core | DataSources | DataSource | WithTransformer]

Contract:

  • pre: “transformer is DataTransformer instance”

  • post: “Returns new DataSource (does NOT modify self)”

Complexity:

2

Parameters:

transformer – DataTransformer instance

Returns:

New DataSource with specified transformer

with_transform_fn(transform_fn: Callable[[DataFrame], DataFrame]) → DataSource[source]

Returns a new datasource instance with an additional transformation step chained AFTER the main data transformer.

Factory method for creating specialized datasources with block-specific transformations. The new transformer is chained after the existing one, preserving the global filter → block transform order.

Hierarchy:

[Core | DataSources | DataSource | WithTransform]

Relates-to:

  • motivated_by: “v0.15.0: Block-specific data transformations”

  • implements: “method: ‘with_transform_fn’”

  • uses: [“class: ‘ChainedTransformer’”]

Rationale:

“Immutable pattern creates specialized clone without modifying original”

Contract:

  • pre: “transform_fn is callable: DataFrame → DataFrame”

  • post: “Returns new DataSource with chained transformer”

  • invariant: “Original datasource unchanged (immutable)”

  • cache: “New datasource has independent cache keys”

Complexity:

4

Decision_cache:

“Use ChainedTransformer for global filter → block transform pipeline”

Parameters:

transform_fn – Function that transforms a DataFrame. Signature: lambda df: df (no params needed) Examples: - lambda df: df.groupby(‘category’).sum() - lambda df: df.pivot_table(…) - lambda df: df.query(“price > 100”)

Returns:

New DataSource instance with chained transformer. Original datasource is unchanged.

Example

>>> # Original datasource with global filter
>>> main_ds = DataSource(
...     data_builder=CSVBuilder("sales.csv"),
...     data_transformer=CategoryFilter()  # Global filter
... )
>>>
>>> # Create specialized datasource for aggregation
>>> agg_ds = main_ds.with_transform_fn(
...     lambda df: df.groupby('category')['sales'].sum().reset_index()
... )
>>>
>>> # Original datasource unchanged, agg_ds has chained transformer
>>> data = agg_ds.get_processed_data({'category': 'Electronics'})
>>> # Flow: Build → CategoryFilter(category='Electronics') → GroupBy Aggregation
get_processed_data(params: Dict[str, Any] | None = None) → DataFrame[source]

Run 2-stage pipeline and return filtered data.

Contract:

  • pre: “params is dict or None”

  • post: “Returns filtered DataFrame”

  • stages: “Build → Filter (2 stages)”

  • invariant: “Stateless (no stored data)”

Complexity:

6

Parameters:

params – Parameters for build + filter

Returns:

Filtered DataFrame from 2-stage pipeline

Raises:

  • DataLoadError – If data loading/building fails

  • CacheError – If cache operations fail (warning only, retries without cache)

  • AsyncSyncMismatchError – If async build_fn used with sync method

class dashboard_lego.core.datasource.DataSource(data_builder: Any | None = None, data_transformer: Any | None = None, param_classifier: Callable[[str], str] | None = None, cache_dir: str | None = None, cache_ttl: int = 300, cache_backend: str | Any | None = None, build_fn: Callable[[Dict[str, Any]], DataFrame] | None = None, transform_fn: Callable[[DataFrame], DataFrame] | None = None, cache_prewarm_params: List[Dict[str, Any]] | None = None, df: DataFrame | None = None, **kwargs)[source]

Bases: object

Base data source with stateless 2-stage pipeline (synchronous only).

Pipeline stages (all via cache): 1. Build (DataBuilder.build) - Load + Process 2. Filter (DataFilter.filter) - Apply filters

NO STORED STATE - data computed fresh each call via cache. NO ABSTRACT METHODS - fully concrete base class. SYNC ONLY - for async support, use AsyncDataSource instead.

Hierarchy:

[Core | DataSources | DataSource]

Relates-to:

  • motivated_by: “v0.15.0: 2-stage pipeline for semantic clarity”

  • implements: “class: ‘DataSource’ stateless (sync only)”

  • uses: [“library: ‘diskcache’”, “class: ‘DataBuilder’”, “class: ‘DataFilter’”]

Rationale:

“2-stage pipeline (Build → Filter) simpler than 3-stage”

Contract:

  • pre: “data_builder and data_filter provided (sync only)”

  • post: “get_processed_data(params) returns filtered data via cache (sync)”

  • invariant: “No stored data attributes, no abstract methods, sync only”

Complexity:

7

Decision_cache:

“Chose 2-stage over 3-stage for semantic clarity. Async methods removed - use AsyncDataSource for async support.”

__init__(data_builder: Any | None = None, data_transformer: Any | None = None, param_classifier: Callable[[str], str] | None = None, cache_dir: str | None = None, cache_ttl: int = 300, cache_backend: str | Any | None = None, build_fn: Callable[[Dict[str, Any]], DataFrame] | None = None, transform_fn: Callable[[DataFrame], DataFrame] | None = None, cache_prewarm_params: List[Dict[str, Any]] | None = None, df: DataFrame | None = None, **kwargs)[source]

Initialize datasource with 2-stage pipeline.

Hierarchy:

[Core | DataSources | DataSource | Initialization]

Relates-to:

  • motivated_by: “v0.15.0: 2-stage pipeline configuration with optional lambda functions”

  • implements: “method: ‘__init__’”

Contract:

  • pre: “df, build_fn, and data_builder are optional (if none provided, uses default DataBuilder)”

  • post: “2-stage pipeline ready with handlers created from df/build_fn/builder if provided, or default DataBuilder”

  • stages: “Build → Transform”

  • priority: “df → build_fn → data_builder → default DataBuilder (first provided wins)”

Parameters:

  • data_builder – DataBuilder for stage 1 (load + process). If None and build_fn or df provided, creates appropriate builder.

  • data_transformer – DataTransformer for stage 2 (filtering/aggregation). If None and transform_fn provided, creates LambdaTransformer.

  • param_classifier – Routes params: ‘build’ or ‘transform’. Default: ‘build__’ → ‘build’, ‘transform__’ → ‘transform’.

  • cache_dir – Directory for disk cache. If None, uses in-memory cache. Ignored if cache_backend is provided.

  • cache_ttl – Time-to-live for cache entries in seconds.

  • cache_backend – Cache backend to use. Can be: - ‘disk’ or None: DiskCacheBackend (default, uses cache_dir) - ‘redis’: RedisCacheBackend (localhost:6379) - ‘memory’: InMemoryCacheBackend - CacheBackend instance: Custom backend

  • build_fn – Optional lambda function for simple data building: Dict[str, Any] → DataFrame. If provided, creates LambdaBuilder automatically. Signature: lambda params: df

  • transform_fn – Optional lambda function for simple data transformation: DataFrame → DataFrame. If provided, creates LambdaTransformer automatically. Signature: lambda df: df

  • cache_prewarm_params – Optional list of parameter dictionaries to prewarm cache during initialization. Each dict will be processed through the 2-stage pipeline to populate cache.

  • df – Optional pandas DataFrame. If provided, automatically creates DfHandler as builder. If none of df/build_fn/data_builder provided, uses default DataBuilder (returns empty DataFrame).

with_builder(builder: Any | Callable) → DataSource[source]

Return new datasource with replaced builder.

Immutable pattern for flexible data pipeline composition.

Hierarchy:

[Core | DataSources | DataSource | WithBuilder]

Contract:

  • pre: “builder is DataBuilder instance”

  • post: “Returns new DataSource (does NOT modify self)”

Complexity:

2

Parameters:

builder – DataBuilder instance

Returns:

New DataSource with specified builder

with_builder_fn(build_fn: Callable[[Dict[str, Any]], DataFrame]) → DataSource[source]

Returns a new datasource instance with a lambda-based builder.

Convenience factory for simple build logic without creating DataBuilder class. Symmetric to with_transform_fn() for consistency.

Hierarchy:

[Core | DataSources | DataSource | WithBuilderFn]

Relates-to:

  • motivated_by: “v0.15.0: Symmetric API with with_transform_fn()”

  • implements: “method: ‘with_builder_fn’”

  • uses: [“class: ‘DataBuilder’”]

Rationale:

“Lambda-based builder for simple cases, avoiding class boilerplate”

Contract:

  • pre: “build_fn is callable: Dict[str, Any] → DataFrame”

  • post: “Returns new DataSource with lambda builder”

  • invariant: “Original datasource unchanged (immutable)”

Complexity:

2

Decision_cache:

“Symmetric with_builder_fn/with_transform_fn API for consistency”

Parameters:

build_fn – Function that builds DataFrame from params. Signature: lambda params: df Examples: - lambda p: pd.read_csv(p[‘file_path’]) - lambda p: generate_sample_data(n=p.get(‘rows’, 100))

Returns:

New DataSource instance with lambda builder. Original datasource is unchanged.

Example

>>> # Create datasource with lambda builder
>>> ds = DataSource().with_builder_fn(
...     lambda params: pd.read_csv('data.csv')
... )
>>>
>>> # Or with params
>>> ds = DataSource().with_builder_fn(
...     lambda params: pd.read_csv(params.get('file', 'default.csv'))
... )
with_transformer(transformer: Any) → DataSource[source]

Return new datasource with replaced transformer.

Immutable pattern for flexible data pipeline composition.

Hierarchy:

[Core | DataSources | DataSource | WithTransformer]

Contract:

  • pre: “transformer is DataTransformer instance”

  • post: “Returns new DataSource (does NOT modify self)”

Complexity:

2

Parameters:

transformer – DataTransformer instance

Returns:

New DataSource with specified transformer

with_transform_fn(transform_fn: Callable[[DataFrame], DataFrame]) → DataSource[source]

Returns a new datasource instance with an additional transformation step chained AFTER the main data transformer.

Factory method for creating specialized datasources with block-specific transformations. The new transformer is chained after the existing one, preserving the global filter → block transform order.

Hierarchy:

[Core | DataSources | DataSource | WithTransform]

Relates-to:

  • motivated_by: “v0.15.0: Block-specific data transformations”

  • implements: “method: ‘with_transform_fn’”

  • uses: [“class: ‘ChainedTransformer’”]

Rationale:

“Immutable pattern creates specialized clone without modifying original”

Contract:

  • pre: “transform_fn is callable: DataFrame → DataFrame”

  • post: “Returns new DataSource with chained transformer”

  • invariant: “Original datasource unchanged (immutable)”

  • cache: “New datasource has independent cache keys”

Complexity:

4

Decision_cache:

“Use ChainedTransformer for global filter → block transform pipeline”

Parameters:

transform_fn – Function that transforms a DataFrame. Signature: lambda df: df (no params needed) Examples: - lambda df: df.groupby(‘category’).sum() - lambda df: df.pivot_table(…) - lambda df: df.query(“price > 100”)

Returns:

New DataSource instance with chained transformer. Original datasource is unchanged.

Example

>>> # Original datasource with global filter
>>> main_ds = DataSource(
...     data_builder=CSVBuilder("sales.csv"),
...     data_transformer=CategoryFilter()  # Global filter
... )
>>>
>>> # Create specialized datasource for aggregation
>>> agg_ds = main_ds.with_transform_fn(
...     lambda df: df.groupby('category')['sales'].sum().reset_index()
... )
>>>
>>> # Original datasource unchanged, agg_ds has chained transformer
>>> data = agg_ds.get_processed_data({'category': 'Electronics'})
>>> # Flow: Build → CategoryFilter(category='Electronics') → GroupBy Aggregation
get_processed_data(params: Dict[str, Any] | None = None) → DataFrame[source]

Run 2-stage pipeline and return filtered data.

Contract:

  • pre: “params is dict or None”

  • post: “Returns filtered DataFrame”

  • stages: “Build → Filter (2 stages)”

  • invariant: “Stateless (no stored data)”

Complexity:

6

Parameters:

params – Parameters for build + filter

Returns:

Filtered DataFrame from 2-stage pipeline

Raises:

  • DataLoadError – If data loading/building fails

  • CacheError – If cache operations fail (warning only, retries without cache)

  • AsyncSyncMismatchError – If async build_fn used with sync method

Data Source Implementations

CSV Data Source

CSV data source with built-in DataBuilder.

hierarchy:

[Core | DataSources | CsvDataSource]

contract:

  • pre: “file_path to CSV provided”

  • post: “CSV loaded and cached”

complexity:

2

class dashboard_lego.core.datasources.csv_source.CsvDataBuilder(file_path: str, read_csv_options: Dict[str, Any] | None = None, **kwargs)[source]

Bases: DataBuilder

DataBuilder for CSV files.

Hierarchy:

[Core | DataSources | CsvDataBuilder]

Contract:

  • pre: “file_path exists”

  • post: “Returns loaded DataFrame”

__init__(file_path: str, read_csv_options: Dict[str, Any] | None = None, **kwargs)[source]

Initialize DataBuilder.

Hierarchy:

[Core | Pipeline | DataBuilder | Init]

Contract:

  • pre: “logger optional”

  • post: “Builder ready”

Parameters:

logger – Optional logger instance

class dashboard_lego.core.datasources.csv_source.CsvDataSource(file_path: str, read_csv_options: Dict[str, Any] | None = None, **kwargs)[source]

Bases: DataSource

CSV data source.

Hierarchy:

[Core | DataSources | CsvDataSource]

Complexity:

2

__init__(file_path: str, read_csv_options: Dict[str, Any] | None = None, **kwargs)[source]

Initialize CSV datasource.

Parameters:

  • file_path – Path to CSV file

  • read_csv_options – Options for pd.read_csv()

Parquet Data Source

Parquet data source with built-in DataBuilder.

hierarchy:

[Core | DataSources | ParquetDataSource]

contract:

  • pre: “file_path to Parquet provided”

  • post: “Parquet loaded and cached”

complexity:

2

class dashboard_lego.core.datasources.parquet_source.ParquetDataBuilder(file_path: str, **kwargs)[source]

Bases: DataBuilder

DataBuilder for Parquet files.

Hierarchy:

[Core | DataSources | ParquetDataBuilder]

Contract:

  • pre: “file_path exists”

  • post: “Returns loaded DataFrame”

__init__(file_path: str, **kwargs)[source]

Initialize DataBuilder.

Hierarchy:

[Core | Pipeline | DataBuilder | Init]

Contract:

  • pre: “logger optional”

  • post: “Builder ready”

Parameters:

logger – Optional logger instance

class dashboard_lego.core.datasources.parquet_source.ParquetDataSource(file_path: str, **kwargs)[source]

Bases: DataSource

Parquet data source.

Hierarchy:

[Core | DataSources | ParquetDataSource]

Complexity:

2

__init__(file_path: str, **kwargs)[source]

Initialize Parquet datasource.

Parameters:

file_path – Path to Parquet file

SQL Data Source

SQL data source with built-in DataBuilder.

hierarchy:

[Core | DataSources | SqlDataSource]

contract:

  • pre: “connection_uri and query provided”

  • post: “SQL data loaded and cached”

complexity:

3

dashboard_lego.core.datasources.sql_source.validate_sql_query(query: str, allowed_statements: list[str] | None = None) → None[source]

Validate SQL query to prevent dangerous operations.

Hierarchy:

[Core | DataSources | Security | ValidateSQL]

Relates-to:

  • motivated_by: “Security: Prevent SQL injection and dangerous operations”

  • implements: “function: ‘validate_sql_query’”

  • uses: [“library: ‘sqlparse’”]

Rationale:

“Only allow SELECT statements by default to prevent data modification.”

Contract:

  • pre: “query is a non-empty string”

  • post: “Raises DataLoadError if query contains dangerous operations”

  • invariant: “Only SELECT statements are allowed by default”

Parameters:

  • query – SQL query string to validate

  • allowed_statements – List of allowed statement types (default: [‘SELECT’])

Raises:

DataLoadError – If query contains dangerous operations or invalid SQL

class dashboard_lego.core.datasources.sql_source.SqlDataBuilder(connection_uri: str, query: str, **kwargs)[source]

Bases: DataBuilder

DataBuilder for SQL databases.

Hierarchy:

[Core | DataSources | SqlDataBuilder]

Contract:

  • pre: “connection_uri and query valid”

  • post: “Returns loaded DataFrame”

__init__(connection_uri: str, query: str, **kwargs)[source]

Initialize DataBuilder.

Hierarchy:

[Core | Pipeline | DataBuilder | Init]

Contract:

  • pre: “logger optional”

  • post: “Builder ready”

Parameters:

logger – Optional logger instance

class dashboard_lego.core.datasources.sql_source.SqlDataSource(connection_uri: str, query: str, allowed_statements: list[str] | None = None, **kwargs)[source]

Bases: DataSource

SQL data source.

Hierarchy:

[Core | DataSources | SqlDataSource]

Complexity:

3

__init__(connection_uri: str, query: str, allowed_statements: list[str] | None = None, **kwargs)[source]

Initialize SQL datasource.

Parameters:

  • connection_uri – SQLAlchemy connection URI

  • query – SQL query to execute

  • allowed_statements – List of allowed statement types (default: [‘SELECT’])

  • **kwargs – Additional arguments passed to DataSource

Raises:

DataLoadError – If query contains dangerous operations or invalid SQL

Dashboard Page

Page package for DashboardPage decomposition.

hierarchy:

[Core | Page | Package]

complexity:

1

class dashboard_lego.core.page.DashboardPage(title: str, blocks: List[List[Any]] | None = None, theme: str = 'https://cdn.jsdelivr.net/npm/bootstrap@5.3.6/dist/css/bootstrap.min.css', navigation: NavigationConfig | None = None, theme_config: ThemeConfig | None = None, sidebar: 'SidebarConfig' | None = None)[source]

Bases: LayoutBuilderMixin, NavigationMixin, SidebarBuilderMixin, CallbacksMixin, ThemeManagerMixin

Orchestrates the assembly of a dashboard page from a list of blocks.

hierarchy:

[Feature | Layout System | Page Modification]

relates-to:

  • motivated_by: “Architectural Conclusion: Provide a flexible grid-based layout system”

  • implements: “class: ‘DashboardPage’”

  • uses: [“interface: ‘BaseBlock’”, “class: ‘StateManager’”]

rationale:

“The page now accepts a nested list structure for layout definition and builds a Bootstrap grid, offering a balance of power and simplicity.”

contract:

  • pre: “blocks must be a list of lists, where each inner item is a BaseBlock or a (BaseBlock, dict) tuple.”

  • post: “A complete Dash layout with a grid structure can be retrieved.”

__init__(title: str, blocks: List[List[Any]] | None = None, theme: str = 'https://cdn.jsdelivr.net/npm/bootstrap@5.3.6/dist/css/bootstrap.min.css', navigation: NavigationConfig | None = None, theme_config: ThemeConfig | None = None, sidebar: 'SidebarConfig' | None = None)[source]

Initializes the DashboardPage, creates a StateManager, and registers all blocks.

Parameters:

  • title – The main title of the dashboard page.

  • blocks –

    A list of lists representing rows. Each item in a row is either a BaseBlock instance or a tuple of (BaseBlock, dict_of_col_props).

    Example:

    [[block1], [(block2, {'width': 8}), (block3, {'width': 4})]]

    If navigation is provided, this parameter is optional.

  • theme – An optional URL to a dash-bootstrap-components theme (e.g., dbc.themes.CYBORG).

  • navigation – Optional NavigationConfig for multi-section dashboard with lazy-loaded content.

  • theme_config – Optional ThemeConfig for global styling customization.

  • sidebar – Optional SidebarConfig for collapsible sidebar with fixed-ID blocks. Sidebar blocks use non-pattern-matched IDs, enabling cross-section State() subscriptions in pattern-matching callbacks.

build_layout() → Component[source]

Assembles the layouts from all blocks into a grid-based page layout.

Supports three layout modes: 1. Sidebar + Navigation: dbc.Offcanvas + multi-section navigation 2. Sidebar + Standard: dbc.Offcanvas + grid layout 3. Standard/Navigation: existing behavior (no sidebar)

CRITICAL: For navigation mode, preload all sections BEFORE building layout to prevent duplicate block creation when combined with sidebar.

Hierarchy:

[Core | Page | BuildLayout]

Relates-to:

  • motivated_by: “Dash callback lifecycle requires all blocks before app.run()”

  • implements: “method: ‘build_layout’ with navigation preload”

  • uses: [“method: ‘_preload_all_section_blocks’”]

Contract:

  • pre: “Page configured with blocks or navigation”

  • post: “Layout built with all blocks created exactly once”

  • invariant: “Navigation sections preloaded before HTML rendering”

  • spec_compliance: “Dash callback registration lifecycle”

Complexity:

5

Decision_cache:

“Preload navigation before layout to prevent duplicate block creation”

Returns:

A Dash component representing the entire page.

export_to_figure(params: Dict[str, Any] | None = None, title: str | None = None) → Figure[source]

Export entire dashboard layout to single Plotly figure.

Combines all chart blocks into a single figure using subplots. Non-chart blocks (metrics, text, controls) are skipped.

Parameters:

  • params – Optional parameters for filtering data in all blocks

  • title – Optional title for the combined figure

Returns:

Single Plotly Figure with all charts in grid layout

Example

>>> page = DashboardPage(title="Sales Dashboard", blocks=layout)
>>> fig = page.export_to_figure(title="Q4 Sales Report")
>>> fig.write_html("dashboard_export.html")

Note

Requires dashboard to be built without navigation (single layout). For navigation dashboards, export sections individually.

class dashboard_lego.core.page.NavigationConfig(sections: List[NavigationSection], position: str = 'left', sidebar_width: int = 3, default_section: int = 0, sidebar_style: Dict[str, Any] | None = None, sidebar_className: str | None = None, content_style: Dict[str, Any] | None = None, content_className: str | None = None, nav_style: Dict[str, Any] | None = None, nav_className: str | None = None, nav_link_style: Dict[str, Any] | None = None, nav_link_className: str | None = None, nav_link_active_style: Dict[str, Any] | None = None, nav_link_active_className: str | None = None)[source]

Bases: object

Configuration for navigation panel in DashboardPage with customizable styling.

hierarchy:

[Feature | Navigation System | NavigationConfig]

relates-to:

  • motivated_by: “PRD: Simplify creation of dashboards with navigation sidebar and customization”

  • implements: “dataclass: ‘NavigationConfig’ with style parameters”

  • uses: [“dataclass: ‘NavigationSection’”]

rationale:

“Encapsulates all navigation settings including style customization in a typed, immutable config object.”

contract:

  • pre: “sections is a non-empty list of NavigationSection instances”

  • post: “Config provides all data needed to render navigation UI with custom styling”

__init__(sections: List[NavigationSection], position: str = 'left', sidebar_width: int = 3, default_section: int = 0, sidebar_style: Dict[str, Any] | None = None, sidebar_className: str | None = None, content_style: Dict[str, Any] | None = None, content_className: str | None = None, nav_style: Dict[str, Any] | None = None, nav_className: str | None = None, nav_link_style: Dict[str, Any] | None = None, nav_link_className: str | None = None, nav_link_active_style: Dict[str, Any] | None = None, nav_link_active_className: str | None = None) → None
content_className: str | None = None
content_style: Dict[str, Any] | None = None
default_section: int = 0
nav_className: str | None = None
nav_style: Dict[str, Any] | None = None
position: str = 'left'
sidebar_className: str | None = None
sidebar_style: Dict[str, Any] | None = None
sidebar_width: int = 3
sections: List[NavigationSection]
class dashboard_lego.core.page.NavigationSection(title: str, block_factory: Callable[[], List[List[Any]]])[source]

Bases: object

Defines a single navigation section with a title and lazy block factory.

hierarchy:

[Feature | Navigation System | NavigationSection]

relates-to:

  • motivated_by: “Architectural Conclusion: Lazy loading of dashboard sections improves performance”

  • implements: “dataclass: ‘NavigationSection’”

  • uses: [“interface: ‘BaseBlock’”]

rationale:

“Uses factory pattern to defer block creation until section is activated.”

contract:

  • pre: “title is a non-empty string, block_factory is a callable returning List[List[Any]]”

  • post: “Section can be rendered on demand via factory invocation”

__init__(title: str, block_factory: Callable[[], List[List[Any]]]) → None
title: str
block_factory: Callable[[], List[List[Any]]]
class dashboard_lego.core.page.DashboardPage(title: str, blocks: List[List[Any]] | None = None, theme: str = 'https://cdn.jsdelivr.net/npm/bootstrap@5.3.6/dist/css/bootstrap.min.css', navigation: NavigationConfig | None = None, theme_config: ThemeConfig | None = None, sidebar: 'SidebarConfig' | None = None)[source]

Bases: LayoutBuilderMixin, NavigationMixin, SidebarBuilderMixin, CallbacksMixin, ThemeManagerMixin

Orchestrates the assembly of a dashboard page from a list of blocks.

hierarchy:

[Feature | Layout System | Page Modification]

relates-to:

  • motivated_by: “Architectural Conclusion: Provide a flexible grid-based layout system”

  • implements: “class: ‘DashboardPage’”

  • uses: [“interface: ‘BaseBlock’”, “class: ‘StateManager’”]

rationale:

“The page now accepts a nested list structure for layout definition and builds a Bootstrap grid, offering a balance of power and simplicity.”

contract:

  • pre: “blocks must be a list of lists, where each inner item is a BaseBlock or a (BaseBlock, dict) tuple.”

  • post: “A complete Dash layout with a grid structure can be retrieved.”

__init__(title: str, blocks: List[List[Any]] | None = None, theme: str = 'https://cdn.jsdelivr.net/npm/bootstrap@5.3.6/dist/css/bootstrap.min.css', navigation: NavigationConfig | None = None, theme_config: ThemeConfig | None = None, sidebar: 'SidebarConfig' | None = None)[source]

Initializes the DashboardPage, creates a StateManager, and registers all blocks.

Parameters:

  • title – The main title of the dashboard page.

  • blocks –

    A list of lists representing rows. Each item in a row is either a BaseBlock instance or a tuple of (BaseBlock, dict_of_col_props).

    Example:

    [[block1], [(block2, {'width': 8}), (block3, {'width': 4})]]

    If navigation is provided, this parameter is optional.

  • theme – An optional URL to a dash-bootstrap-components theme (e.g., dbc.themes.CYBORG).

  • navigation – Optional NavigationConfig for multi-section dashboard with lazy-loaded content.

  • theme_config – Optional ThemeConfig for global styling customization.

  • sidebar – Optional SidebarConfig for collapsible sidebar with fixed-ID blocks. Sidebar blocks use non-pattern-matched IDs, enabling cross-section State() subscriptions in pattern-matching callbacks.

build_layout() → Component[source]

Assembles the layouts from all blocks into a grid-based page layout.

Supports three layout modes: 1. Sidebar + Navigation: dbc.Offcanvas + multi-section navigation 2. Sidebar + Standard: dbc.Offcanvas + grid layout 3. Standard/Navigation: existing behavior (no sidebar)

CRITICAL: For navigation mode, preload all sections BEFORE building layout to prevent duplicate block creation when combined with sidebar.

Hierarchy:

[Core | Page | BuildLayout]

Relates-to:

  • motivated_by: “Dash callback lifecycle requires all blocks before app.run()”

  • implements: “method: ‘build_layout’ with navigation preload”

  • uses: [“method: ‘_preload_all_section_blocks’”]

Contract:

  • pre: “Page configured with blocks or navigation”

  • post: “Layout built with all blocks created exactly once”

  • invariant: “Navigation sections preloaded before HTML rendering”

  • spec_compliance: “Dash callback registration lifecycle”

Complexity:

5

Decision_cache:

“Preload navigation before layout to prevent duplicate block creation”

Returns:

A Dash component representing the entire page.

export_to_figure(params: Dict[str, Any] | None = None, title: str | None = None) → Figure[source]

Export entire dashboard layout to single Plotly figure.

Combines all chart blocks into a single figure using subplots. Non-chart blocks (metrics, text, controls) are skipped.

Parameters:

  • params – Optional parameters for filtering data in all blocks

  • title – Optional title for the combined figure

Returns:

Single Plotly Figure with all charts in grid layout

Example

>>> page = DashboardPage(title="Sales Dashboard", blocks=layout)
>>> fig = page.export_to_figure(title="Q4 Sales Report")
>>> fig.write_html("dashboard_export.html")

Note

Requires dashboard to be built without navigation (single layout). For navigation dashboards, export sections individually.

State Management

This module defines the StateManager for handling interactivity between blocks.

class dashboard_lego.core.state.StateManager[source]

Bases: object

Manages the state dependencies and generates callbacks for a dashboard page.

This class acts as a central registry for components that provide state (publishers) and components that consume state (subscribers). It builds a dependency graph and will be responsible for generating the necessary Dash callbacks to link them.

hierarchy:

[Feature | Global Interactivity | StateManager Design]

relates-to:

  • motivated_by: “Architectural Conclusion: Decouple state management from UI components using a Pub/Sub model”

  • implements: “class: ‘StateManager’”

  • uses: []

rationale:

“Chosen a graph-like dictionary structure to store state dependencies. This provides a good balance of implementation simplicity and ease of traversal for callback generation.”

contract:

  • pre: “All state IDs must be unique across the application.”

  • post: “The manager holds a complete dependency graph of the page’s interactive components.”

__init__()[source]

Initializes the StateManager.

The internal dependency_graph will store the relationships.

Example:

{
    'selected_date_range': {
        'publisher': {
            'component_id': 'global-date-picker',
            'component_prop': 'value'
        },
        'subscribers': [
            {
                'component_id': 'sales-trend-graph',
                'component_prop': 'figure',
                'callback_fn': '<function_ref>'
            },
            {
                'component_id': 'kpi-block-container',
                'component_prop': 'children',
                'callback_fn': '<function_ref>'
            }
        ]
    }
}
register_publisher(state_id: str, component_id: str, component_prop: str, dep_param_name: str | None = None)[source]

Registers a component property as a provider of a certain state.

Parameters:

  • state_id – The unique identifier for the state (e.g., ‘selected_date_range’).

  • component_id – The ID of the Dash component that publishes the state.

  • component_prop – The property of the component that holds the state (e.g., ‘value’).

register_subscriber(state_id: str, component_id: str, component_prop: str, callback_fn: Callable)[source]

Registers a component property as a consumer of a certain state.

Parameters:

  • state_id – The unique identifier for the state to subscribe to.

  • component_id – The ID of the Dash component that consumes the state.

  • component_prop – The property of the component to be updated (e.g., ‘figure’).

  • callback_fn – The function to call to generate the new property value.

get_initial_publisher_values() → Dict[str, Any][source]

Get initial values of all registered publishers.

Hierarchy:

[Feature | Initial State Sync | StateManager]

Relates-to:

  • motivated_by: “Blocks need initial state values before rendering”

  • implements: “method: ‘get_initial_publisher_values’”

Contract:

  • pre: “Publishers are registered”

  • post: “Returns {state_id: None or value} for all states”

Returns:

Dict mapping state_id to initial value (None if not available)

generate_callbacks(app: Any, blocks: List[Any] = None)[source]

Traverses the dependency graph and registers all necessary callbacks with the Dash app.

Parameters:

  • app – Dash app instance

  • blocks – List of blocks (to check for controls)

This method now supports multi-state subscriptions by grouping subscribers by their output target and creating callbacks with multiple Input sources.

Hierarchy:

[Feature | Multi-State Subscription | StateManager]

Relates-to:

  • motivated_by: “Bug Fix: Support subscribing to multiple states”

  • implements: “method: ‘generate_callbacks’ with multi-input support”

Rationale:

“Group subscriptions by output target to create one callback per subscriber with multiple inputs, avoiding duplicate output errors.”

Contract:

  • pre: “Dependency graph is populated with publishers and subscribers.”

  • post: “One callback per unique output target with all its input states.”

Parameters:

app – The Dash app instance.

bind_callbacks(app: Any, blocks: List[Any])[source]

Registers one callback per block instead of per state.

Hierarchy:

[Architecture | Block-centric Callbacks | StateManager]

Relates-to:

  • motivated_by: “Architectural Conclusion: Block-centric callbacks improve performance and maintainability by reducing callback complexity”

  • implements: “method: ‘bind_callbacks’”

  • uses: [“method: ‘output_target’”, “method: ‘list_control_inputs’”]

Rationale:

“Each block gets exactly one callback that updates its output target.”

Contract:

  • pre: “Blocks must have output_target() and list_control_inputs() methods.”

  • post: “Each block has exactly one callback registered with Dash.”

Parameters:

  • app – The Dash app instance.

  • blocks – List of blocks to register callbacks for.

clear_registered_outputs() → None[source]

Clear registered output cache.

Use when blocks are being re-rendered (e.g., section removal).

Hierarchy:

[Feature | Callback Management | StateManager]

Relates-to:

  • motivated_by: “Navigation lazy-loading needs to clear callback cache”

  • implements: “method: ‘clear_registered_outputs’”

Contract:

  • pre: “No active callbacks in Dash”

  • post: “_registered_outputs is empty”

class dashboard_lego.core.state.StateManager[source]

Bases: object

Manages the state dependencies and generates callbacks for a dashboard page.

This class acts as a central registry for components that provide state (publishers) and components that consume state (subscribers). It builds a dependency graph and will be responsible for generating the necessary Dash callbacks to link them.

hierarchy:

[Feature | Global Interactivity | StateManager Design]

relates-to:

  • motivated_by: “Architectural Conclusion: Decouple state management from UI components using a Pub/Sub model”

  • implements: “class: ‘StateManager’”

  • uses: []

rationale:

“Chosen a graph-like dictionary structure to store state dependencies. This provides a good balance of implementation simplicity and ease of traversal for callback generation.”

contract:

  • pre: “All state IDs must be unique across the application.”

  • post: “The manager holds a complete dependency graph of the page’s interactive components.”

__init__()[source]

Initializes the StateManager.

The internal dependency_graph will store the relationships.

Example:

{
    'selected_date_range': {
        'publisher': {
            'component_id': 'global-date-picker',
            'component_prop': 'value'
        },
        'subscribers': [
            {
                'component_id': 'sales-trend-graph',
                'component_prop': 'figure',
                'callback_fn': '<function_ref>'
            },
            {
                'component_id': 'kpi-block-container',
                'component_prop': 'children',
                'callback_fn': '<function_ref>'
            }
        ]
    }
}
register_publisher(state_id: str, component_id: str, component_prop: str, dep_param_name: str | None = None)[source]

Registers a component property as a provider of a certain state.

Parameters:

  • state_id – The unique identifier for the state (e.g., ‘selected_date_range’).

  • component_id – The ID of the Dash component that publishes the state.

  • component_prop – The property of the component that holds the state (e.g., ‘value’).

register_subscriber(state_id: str, component_id: str, component_prop: str, callback_fn: Callable)[source]

Registers a component property as a consumer of a certain state.

Parameters:

  • state_id – The unique identifier for the state to subscribe to.

  • component_id – The ID of the Dash component that consumes the state.

  • component_prop – The property of the component to be updated (e.g., ‘figure’).

  • callback_fn – The function to call to generate the new property value.

get_initial_publisher_values() → Dict[str, Any][source]

Get initial values of all registered publishers.

Hierarchy:

[Feature | Initial State Sync | StateManager]

Relates-to:

  • motivated_by: “Blocks need initial state values before rendering”

  • implements: “method: ‘get_initial_publisher_values’”

Contract:

  • pre: “Publishers are registered”

  • post: “Returns {state_id: None or value} for all states”

Returns:

Dict mapping state_id to initial value (None if not available)

generate_callbacks(app: Any, blocks: List[Any] = None)[source]

Traverses the dependency graph and registers all necessary callbacks with the Dash app.

Parameters:

  • app – Dash app instance

  • blocks – List of blocks (to check for controls)

This method now supports multi-state subscriptions by grouping subscribers by their output target and creating callbacks with multiple Input sources.

Hierarchy:

[Feature | Multi-State Subscription | StateManager]

Relates-to:

  • motivated_by: “Bug Fix: Support subscribing to multiple states”

  • implements: “method: ‘generate_callbacks’ with multi-input support”

Rationale:

“Group subscriptions by output target to create one callback per subscriber with multiple inputs, avoiding duplicate output errors.”

Contract:

  • pre: “Dependency graph is populated with publishers and subscribers.”

  • post: “One callback per unique output target with all its input states.”

Parameters:

app – The Dash app instance.

bind_callbacks(app: Any, blocks: List[Any])[source]

Registers one callback per block instead of per state.

Hierarchy:

[Architecture | Block-centric Callbacks | StateManager]

Relates-to:

  • motivated_by: “Architectural Conclusion: Block-centric callbacks improve performance and maintainability by reducing callback complexity”

  • implements: “method: ‘bind_callbacks’”

  • uses: [“method: ‘output_target’”, “method: ‘list_control_inputs’”]

Rationale:

“Each block gets exactly one callback that updates its output target.”

Contract:

  • pre: “Blocks must have output_target() and list_control_inputs() methods.”

  • post: “Each block has exactly one callback registered with Dash.”

Parameters:

  • app – The Dash app instance.

  • blocks – List of blocks to register callbacks for.

clear_registered_outputs() → None[source]

Clear registered output cache.

Use when blocks are being re-rendered (e.g., section removal).

Hierarchy:

[Feature | Callback Management | StateManager]

Relates-to:

  • motivated_by: “Navigation lazy-loading needs to clear callback cache”

  • implements: “method: ‘clear_registered_outputs’”

Contract:

  • pre: “No active callbacks in Dash”

  • post: “_registered_outputs is empty”

Chart Context

Chart context data structure for unified chart generation.

This module defines the ChartContext dataclass that provides a standardized interface for chart generators, containing all necessary context information including datasource, controls, and logger.

class dashboard_lego.core.chart_context.ChartContext(datasource: DataSource, controls: Dict[str, Any], logger: Any = None)[source]

Bases: object

Context object containing all necessary information for chart generation.

This dataclass provides a unified interface for chart generators, encapsulating the datasource, control values, and logger in a single immutable object.

hierarchy:

[Architecture | ChartContext | ChartContext]

relates-to:

  • motivated_by: “Architectural Conclusion: Unified chart generator signatures enable consistent chart creation across different block types”

  • implements: “datatype: ‘ChartContext’”

  • uses: [“interface: ‘DataSource’”, “module: ‘utils.logger’”]

rationale:

“Chose immutable dataclass for thread safety and clear interface contract.”

contract:

  • pre: “Valid datasource and controls dictionary must be provided.”

  • post: “Context object contains all necessary data for chart generation.”

datasource: DataSource
controls: Dict[str, Any]
logger: Any = None
__post_init__()[source]

Initialize logger if not provided.

__init__(datasource: DataSource, controls: Dict[str, Any], logger: Any = None) → None
class dashboard_lego.core.chart_context.ChartContext(datasource: DataSource, controls: Dict[str, Any], logger: Any = None)[source]

Bases: object

Context object containing all necessary information for chart generation.

This dataclass provides a unified interface for chart generators, encapsulating the datasource, control values, and logger in a single immutable object.

hierarchy:

[Architecture | ChartContext | ChartContext]

relates-to:

  • motivated_by: “Architectural Conclusion: Unified chart generator signatures enable consistent chart creation across different block types”

  • implements: “datatype: ‘ChartContext’”

  • uses: [“interface: ‘DataSource’”, “module: ‘utils.logger’”]

rationale:

“Chose immutable dataclass for thread safety and clear interface contract.”

contract:

  • pre: “Valid datasource and controls dictionary must be provided.”

  • post: “Context object contains all necessary data for chart generation.”

datasource: DataSource
controls: Dict[str, Any]
logger: Any = None
__post_init__()[source]

Initialize logger if not provided.

__init__(datasource: DataSource, controls: Dict[str, Any], logger: Any = None) → None