Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _adapter_heartbeat method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

server_v2.py

Source:server_v2.py

...333                data=init_err.__dict__,334            )335        )336    loop = asyncio.get_running_loop()337    project.heartbeat = loop.create_task(_adapter_heartbeat(project))338    return OsmosisRegisterResult(added=x_dbt_project, projects=dbt.registered_projects())339@app.post(340    "/unregister",341    responses={status.HTTP_404_NOT_FOUND: {"model": OsmosisErrorContainer}},342)343async def unregister(344    response: Response,345    x_dbt_project: str = Header(),346) -> Union[OsmosisResetResult, OsmosisErrorContainer]:347    """Unregister a project. This drop a project from memory"""348    dbt: DbtProjectContainer = app.state.dbt_project_container349    project = dbt.get_project(x_dbt_project)350    if project is None:351        response.status_code = status.HTTP_404_NOT_FOUND352        return OsmosisErrorContainer(353            error=OsmosisError(354                code=OsmosisErrorCode.ProjectNotRegistered,355                message="Project is not registered. Make a POST request to the /register endpoint first to register a runner",356                data={"registered_projects": dbt.registered_projects()},357            )358        )359    project.heartbeat.cancel()360    dbt.drop_project(project)361    return OsmosisUnregisterResult(removed=project, projects=dbt.registered_projects())362@app.get("/health")363async def health_check(364    x_dbt_project: str = Header(default=DEFAULT),365) -> dict:366    """Checks if the server is running and accepting requests367    TODO: We will likely iterate on this, it is mostly for368    kubernetes liveness probes"""369    dbt: DbtProjectContainer = app.state.dbt_project_container370    if x_dbt_project == DEFAULT:371        project = dbt.get_default_project()372    else:373        project = dbt.get_project(x_dbt_project)374    return {375        "result": {376            "status": "ready",377            **(378                {379                    "project_name": project.config.project_name,380                    "target_name": project.config.target_name,381                    "profile_name": project.config.profile_name,382                    "logs": project.config.log_path,383                    "runner_parse_iteration": project._version,384                    "adapter_ready": project.adapter_probe(),385                }386                if project is not None387                else {}388            ),389            "timestamp": datetime.datetime.utcnow(),390            "error": None,391        },392        "dbt-osmosis-server": __name__,393    }394async def _adapter_heartbeat(runner: DbtProject):395    """Equivalent of a keepalive for adapters such as Snowflake"""396    await asyncio.sleep(60 * 30)397    while runner.adapter_probe():398        await asyncio.sleep(60 * 30)399def run_server(host="localhost", port=8581):400    uvicorn.run(401        "dbt_osmosis.core.server_v2:app",402        host=host,403        port=port,404        log_level="info",405        reload=False,406        workers=1,407    )408def test_server():...

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.