Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _recurse_load_queries(path):
queries = Queries()
for p in path.iterdir():
if p.is_file() and p.suffix != ".sql":
continue
elif p.is_file() and p.suffix == ".sql":
for query_name, fn in load_queries_from_file(p, driver_adapter):
queries.add_query(query_name, fn)
elif p.is_dir():
child_name = p.relative_to(dir_path).name
child_queries = _recurse_load_queries(p)
queries.add_child_queries(child_name, child_queries)
else:
# This should be practically unreachable.
raise SQLLoadException(f"The path must be a directory or file, got {p}")
return queries
import sqlite3
import aiosql
queries = aiosql.from_path("./greetings.sql", driver_name="sqlite3")
queries2 = aiosql.from_path("./sql_dir", driver_name="sqlite3")
"""
path = Path(sql_path)
if not path.exists():
raise SQLLoadException(f"File does not exist: {path}")
driver_adapter = get_driver_adapter(driver_name)
if path.is_file():
return Queries(load_queries_from_file(path, driver_adapter))
elif path.is_dir():
return load_queries_from_dir_path(path, driver_adapter)
else:
raise SQLLoadException(f"The sql_path must be a directory or file, got {sql_path}")
-- Get all the greetings in the database
select * from greetings;
-- name: get-users-by-username
-- Get all the users from the database,
-- and return it as a dict
select * from users where username =:username;
\"""
queries = aiosql.from_str(sql_text, db_driver="sqlite3")
queries.get_all_greetings(conn)
queries.get_users_by_username(conn, username="willvaughn")
"""
driver_adapter = get_driver_adapter(driver_name)
return Queries(load_queries_from_sql(sql, driver_adapter))