dlt.helpers.studio.utils
resolve_studio_config
def resolve_studio_config(p: dlt.Pipeline) -> StudioConfiguration
Resolve the studio configuration
get_local_pipelines
def get_local_pipelines(
pipelines_dir: str = None,
sort_by_trace: bool = True) -> Tuple[str, List[Dict[str, Any]]]
Get the local pipelines directory and the list of pipeline names in it.
Arguments:
pipelines_dirstr, optional - The local pipelines directory. Defaults to get_dlt_pipelines_dir().sort_by_tracebool, optional - Whether to sort the pipelines by the latet timestamp of trace. Defaults to True.
Returns:
Tuple[str, List[str]]: The local pipelines directory and the list of pipeline names in it.
get_pipeline
def get_pipeline(pipeline_name: str, pipelines_dir: str) -> dlt.Pipeline
Get a pipeline by name.
Arguments:
pipeline_namestr - The name of the pipeline to get.
Returns:
dlt.Pipeline- The pipeline.
pipeline_details
def pipeline_details(pipeline: dlt.Pipeline) -> List[Dict[str, Any]]
Get the details of a pipeline.
create_table_list
def create_table_list(c: StudioConfiguration,
pipeline: dlt.Pipeline,
show_internals: bool = False,
show_child_tables: bool = True,
show_row_counts: bool = False) -> List[Dict[str, str]]
Create a list of tables for the pipeline.
Arguments:
pipeline_namestr - The name of the pipeline to create the table list for.
create_column_list
def create_column_list(
c: StudioConfiguration,
pipeline: dlt.Pipeline,
table_name: str,
show_internals: bool = False,
show_type_hints: bool = True,
show_other_hints: bool = False,
show_custom_hints: bool = False) -> List[Dict[str, Any]]
Create a list of columns for a table.
Arguments:
pipeline_namestr - The name of the pipeline to create the column list for.table_namestr - The name of the table to create the column list for.
clear_query_cache
def clear_query_cache(pipeline: dlt.Pipeline) -> None
Clear the query cache and history
get_query_result
@functools.cache
def get_query_result(pipeline: dlt.Pipeline, query: str) -> pd.DataFrame
Get the result of a query.
get_row_counts
@functools.cache
def get_row_counts(pipeline: dlt.Pipeline,
load_id: str = None) -> Dict[str, int]
Get the row counts for a pipeline.
Arguments:
pipelinedlt.Pipeline - The pipeline to get the row counts for.load_idstr - The load id to get the row counts for.
get_loads
@functools.cache
def get_loads(c: StudioConfiguration,
pipeline: dlt.Pipeline,
limit: int = 100) -> Any
Get the loads of a pipeline.
get_schema_by_version
@functools.cache
def get_schema_by_version(pipeline: dlt.Pipeline, version_hash: str) -> Schema
Get the schema version of a pipeline.
trace_overview
def trace_overview(c: StudioConfiguration,
trace: Dict[str, Any]) -> List[Dict[str, Any]]
Get the overview of a trace.
trace_execution_context
def trace_execution_context(c: StudioConfiguration,
trace: Dict[str, Any]) -> List[Dict[str, Any]]
Get the execution context of a trace.
trace_steps_overview
def trace_steps_overview(c: StudioConfiguration,
trace: Dict[str, Any]) -> List[Dict[str, Any]]
Get the steps overview of a trace.
trace_resolved_config_values
def trace_resolved_config_values(
c: StudioConfiguration, trace: Dict[str, Any]) -> List[Dict[str, Any]]
Get the resolved config values of a trace.
trace_step_details
def trace_step_details(c: StudioConfiguration, trace: Dict[str, Any],
step_id: str) -> List[Dict[str, Any]]
Get the details of a step.
style_cell
def style_cell(row_id: str, name: str, __: Any) -> Dict[str, str]
Style a cell in a table.
Arguments:
row_idstr - The id of the row.namestr - The name of the column.__Any - The value of the cell.
Returns:
Dict[str, str]: The css style of the cell.