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 ð¬")...
Check out the latest blogs from LambdaTest on this topic:
Building a website is all about keeping the user experience in mind. Ultimately, it’s about providing visitors with a mind-blowing experience so they’ll keep coming back. One way to ensure visitors have a great time on your site is to add some eye-catching text or image animations.
When most firms employed a waterfall development model, it was widely joked about in the industry that Google kept its products in beta forever. Google has been a pioneer in making the case for in-production testing. Traditionally, before a build could go live, a tester was responsible for testing all scenarios, both defined and extempore, in a testing environment. However, this concept is evolving on multiple fronts today. For example, the tester is no longer testing alone. Developers, designers, build engineers, other stakeholders, and end users, both inside and outside the product team, are testing the product and providing feedback.
Entering the world of testers, one question started to formulate in my mind: “what is the reason that bugs happen?”.
In 2007, Steve Jobs launched the first iPhone, which revolutionized the world. But because of that, many businesses dealt with the problem of changing the layout of websites from desktop to mobile by delivering completely different mobile-compatible websites under the subdomain of ‘m’ (e.g., https://m.facebook.com). And we were all trying to figure out how to work in this new world of contending with mobile and desktop screen sizes.
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!