How to use run_sql method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

difficulty_repository.py

Source:difficulty_repository.py

...4#save5def save(difficulty):6    sql = "INSERT INTO difficulties (level) VALUES (%s) RETURNING id"7    values = [difficulty.level]8    results = run_sql(sql, values)9    id = results[0]['id']10    difficulty.id = id11    return difficulty12#select13def select(id):14    sql = "SELECT * FROM difficulties WHERE id = %s"15    values = [id]16    result = run_sql(sql, values)[0]17    difficulty = Difficulty(result["level"], result["id"])18    return difficulty19#select-all20def select_all():21    all_difficulties = []22    sql = "SELECT * FROM difficulties"23    results = run_sql(sql)24    25    for result in results:26        difficulty = Difficulty(result["level"], result["id"])27        all_difficulties.append(difficulty)28    return all_difficulties29#delete30def delete(id):31    sql = "DELETE FROM difficulties WHERE id = %s"32    values = [id]33    run_sql(sql, values)34#delete-all35def delete_all():36    sql = "DELETE FROM difficulties"37    run_sql(sql)38#update39def update(topic):40    sql = "UPDATE difficulties SET (level) = (%s) WHERE id = %s"41    values = [difficulty.level, difficulty.id]...

user_topic_repository.py

Source:user_topic_repository.py

...4#save5def save(user_topic):6    sql = "INSERT INTO user_topics (name) VALUES (%s) RETURNING id"7    values = [user_topic.name]8    results = run_sql(sql, values)9    id = results[0]['id']10    user_topic.id = id11    return user_topic12#select13def select(id):14    sql = "SELECT * FROM user_topics WHERE id = %s"15    values = [id]16    result = run_sql(sql, values)[0]17    user_topic = UserTopic(result["name"], result["id"])18    return user_topic19#select-all20def select_all():21    user_topics = []22    sql = "SELECT * FROM user_topics"23    results = run_sql(sql)24    for result in results:25        user_topic = UserTopic(result["name"], result["id"])26        user_topics.append(user_topic)27    return user_topics28#delete29def delete(id):30    sql = "DELETE FROM user_topics WHERE id = %s"31    values = [id]32    run_sql(sql, values)33#delete-all34def delete_all():35    sql = "DELETE FROM user_topics"36    run_sql(sql)37#update38def update(user_topic):39    sql = "UPDATE user_topics SET (name) = (%s) WHERE id = %s"40    values = [user_topic.name, user_topic.id]...

topic_repository.py

Source:topic_repository.py

...4#save5def save(topic):6    sql = "INSERT INTO topics (name) VALUES (%s) RETURNING id"7    values = [topic.name]8    results = run_sql(sql, values)9    id = results[0]['id']10    topic.id = id11    return topic12#select13def select(id):14    sql = "SELECT * FROM topics WHERE id = %s"15    values = [id]16    result = run_sql(sql, values)[0]17    topic = Topic(result["name"], result["id"])18    return topic19#select-all20def select_all():21    topics = []22    sql = "SELECT * FROM topics"23    results = run_sql(sql)24    for result in results:25        topic = Topic(result["name"], result["id"])26        topics.append(topic)27    return topics28#delete29def delete(id):30    sql = "DELETE FROM topics WHERE id = %s"31    values = [id]32    run_sql(sql, values)33#delete-all34def delete_all():35    sql = "DELETE FROM topics"36    run_sql(sql)37#update38def update(topic):39    sql = "UPDATE topics SET (name) = (%s) WHERE id = %s"40    values = [topic.name, topic.id]...

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.