How to use inject_dbt method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

app.py

Source:app.py

...119"""Run pandas profiler on test bench result set"""120PIVOT_LAYOUT = "PIVOT_LAYOUT"121"""Pivot the editor layout from side-by-side to top-bottom"""122state.setdefault(PIVOT_LAYOUT, False)123def inject_dbt(change_target: Optional[str] = None):124    """Parse dbt project and load context var"""125    if DBT not in state or change_target:126        dbt_ctx = DbtProject(127            project_dir=state[PROJ_DIR],128            profiles_dir=state[PROF_DIR],129            target=change_target,130        )131    else:132        dbt_ctx: DbtProject = state[DBT]133        dbt_ctx.rebuild_dbt_manifest(reset=True)134    state[DBT] = dbt_ctx135    return True136if DBT not in state:137    inject_dbt()138ctx: DbtProject = state[DBT]139TARGET_PROFILE = "TARGET_PROFILE"140"""Target profile for dbt to execute against"""141state.setdefault(TARGET_PROFILE, ctx.config.target_name)142def toggle_viewer() -> None:143    state[PIVOT_LAYOUT] = not state[PIVOT_LAYOUT]144# @st.cache145def compile_sql(sql: str) -> str:146    try:147        return ctx.compile_sql(sql).compiled_sql148    except CompilationException:149        return None150def run_query(sql: str, limit: int = 2000) -> None:151    try:152        result = ctx.execute_sql(f"select * from ({sql}) as __all_data limit {limit}")153    except DatabaseException as error:154        state[SQL_QUERY_STATE] = "error"155        state[SQL_ADAPTER_RESP] = str(error)156    else:157        output = [OrderedDict(zip(result.table.column_names, row)) for row in result.table.rows]158        state[SQL_RESULT] = pd.DataFrame(output)159        state[SQL_ADAPTER_RESP] = result.adapter_response160        state[SQL_QUERY_STATE] = "success"161@st.cache162def convert_df_to_csv(dataframe: pd.DataFrame):163    return dataframe.to_csv().encode("utf-8")164@st.cache(165    hash_funcs={166        pandas_profiling.report.presentation.core.container.Container: lambda _: state[167            COMPILED_SQL168        ],169        pandas_profiling.report.presentation.core.html.HTML: lambda _: state[COMPILED_SQL],170    },171    allow_output_mutation=True,172)173def build_profile_report(174    dataframe: pd.DataFrame, minimal: bool = True175) -> pandas_profiling.ProfileReport:176    return dataframe.profile_report(minimal=minimal)177@st.cache(178    hash_funcs={179        pandas_profiling.report.presentation.core.container.Container: lambda _: state[180            COMPILED_SQL181        ],182        pandas_profiling.report.presentation.core.html.HTML: lambda _: state[COMPILED_SQL],183    },184    allow_output_mutation=True,185)186def convert_profile_report_to_html(profile: pandas_profiling.ProfileReport) -> str:187    return profile.to_html()188st.title("dbt-osmosis ð")189st.sidebar.header("Profiles")190st.sidebar.write(191    "Select a profile used for materializing, compiling, and testing models. Can be updated at any time."192)193state[TARGET_PROFILE] = st.sidebar.radio(194    f"Loaded profiles from {ctx.config.profile_name}",195    [target for target in state[RAW_PROFILES][ctx.config.profile_name].get("outputs", [])],196    key=PROFILE_SELECTOR,197)198st.sidebar.markdown(f"Current Target: **{state[TARGET_PROFILE]}**")199st.sidebar.write("")200st.sidebar.write("Utility")201# st.sidebar.button("Reload dbt project", key=DBT_DO_RELOAD)202st.sidebar.caption(203    "Refresh the page to reparse dbt. This is useful if any updated models or macros in your physical project \204    on disk have changed and are not yet reflected in the workbench as refable or updated."205)206st.sidebar.write("")207st.sidebar.selectbox("Editor Theme", THEMES, index=8, key=THEME_PICKER)208st.sidebar.selectbox("Editor Language", DIALECTS, key=DIALECT_PICKER)209# IDE LAYOUT210notificationContainer = st.empty()211descriptionContainer = st.container()212compileOptionContainer = st.container()213ideContainer = st.container()214descriptionContainer.markdown(215    """216Welcome to the [dbt-osmosis](https://github.com/z3z1ma/dbt-osmosis) workbench ð. 217The workbench serves as a no fuss way to spin up 218an environment where you can very quickly iterate on dbt models. In an ideal flow, a developer219can spin up the workbench and use it as a _complement_ to their IDE, not a replacement. This means220copying and pasting over a model you are really digging into ð§âð» OR it is just as valid to use 221the workbench as a scratchpad ð·ââï¸. In a full day of development, you may never spin down the workbench.222Refreshing the page is enough to reparse the physical dbt project on disk. The instantaneous feedback223rarely experienced with jinja + ability to execute the SQL both synergize to supercharge â¡ï¸ productivity!224"""225)226if not state[PIVOT_LAYOUT]:227    idePart1, idePart2 = ideContainer.columns(2)228else:229    idePart1 = ideContainer.container()230    idePart2 = ideContainer.container()231compileOptionContainer.write("")232compileOpt1, compileOpt2 = compileOptionContainer.columns(2)233auto_update = compileOpt1.checkbox("Dynamic Compilation", key=DYNAMIC_COMPILATION, value=True)234if auto_update:235    compileOpt1.caption("ð Compiling SQL on change")236else:237    compileOpt1.caption("ð Compiling SQL with control + enter")238compileOpt2.button("Pivot Layout", on_click=toggle_viewer)239with idePart1:240    state[RAW_SQL] = st_ace(241        value=state[RAW_SQL],242        theme=state[THEME_PICKER],243        language=state[DIALECT_PICKER],244        auto_update=auto_update,245        key=f"AceEditor",246        max_lines=35,247        min_lines=20,248        height=500,249    )250with idePart2:251    with st.expander("ð Compiled SQL", expanded=True):252        st.code(253            state[COMPILED_SQL]254            if state[COMPILED_SQL]255            else " --> Invalid Jinja, awaiting model to become valid",256            language="sql",257        )258if compile_sql(state[RAW_SQL]) != state[COMPILED_SQL]:259    state[COMPILED_SQL] = compile_sql(state[RAW_SQL])260    st.experimental_rerun()  # This eager re-run speeds up the app261if ctx.config.target_name != state[TARGET_PROFILE]:  # or state[DBT_DO_RELOAD]:262    print("Reloading dbt project...")263    with notificationContainer:264        ctx.config.target_name = state[TARGET_PROFILE]265        ctx.config.target_name = state[TARGET_PROFILE]266        with st.spinner("Reloading dbt... âï¸"):267            inject_dbt(state[TARGET_PROFILE])268            # state[RAW_SQL] += " "269            state[COMPILED_SQL] = compile_sql(state[RAW_SQL])270    st.experimental_rerun()271# TEST LAYOUT272testHeaderContainer = st.container()273test_column_1, _, test_column_2 = st.columns([1, 2, 1])274testContainer = st.container()275testContainerViewer = testContainer.expander("Result Viewer ð", expanded=True)276test_view_1, _, test_view_2 = testContainerViewer.columns([1, 2, 1])277downloadBtnContainer, profileBtnContainer, profileOptContainer = st.columns([1, 1, 3])278profilerContainer = st.container()279with testHeaderContainer:280    st.write("")281    st.subheader("Osmosis Query Result Inspector ð¬")...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.