# credo:disable-for-this-file Credo.Check.Refactor.LongQuoteBlocks defmodule SqlKit do @moduledoc """ Execute raw SQL in strings or .sql files, get maps and structs back. SqlKit provides two ways to execute SQL with automatic result transformation: ## Direct SQL Execution Execute SQL strings directly with any Ecto repo: SqlKit.query_all(MyApp.Repo, "SELECT * FROM users WHERE age > $1", [21]) # => [%{id: 1, name: "Alice", age: 30}, ...] SqlKit.query_one!(MyApp.Repo, "SELECT * FROM users WHERE id = $1", [1], as: User) # => %User{id: 1, name: "Alice"} ## File-Based SQL For larger queries, keep SQL in dedicated files with compile-time embedding: defmodule MyApp.Reports.SQL do use SqlKit, otp_app: :my_app, repo: MyApp.Repo, dirname: "reports", files: ["stats.sql"] end MyApp.Reports.SQL.query_one!("stats.sql", [report_id]) ### DuckDB File-Based SQL For DuckDB, use the `:backend` option instead of `:repo`. The pool must be started in your supervision tree with the database configuration: # In your application.ex supervision tree: children = [ {SqlKit.DuckDB.Pool, name: MyApp.AnalyticsPool, database: "priv/analytics.duckdb", pool_size: 4} ] # Then define your SQL module: defmodule MyApp.Analytics.SQL do use SqlKit, otp_app: :my_app, backend: {:duckdb, pool: MyApp.AnalyticsPool}, dirname: "analytics", files: ["daily_summary.sql"] end MyApp.Analytics.SQL.query_all("daily_summary.sql", [~D[2024-01-01]]) ## Supported Databases Any Ecto adapter returning a result map containing rows and columns should work. The test suite covers the following adapters: | Database | Ecto Adapter | Driver | |------------|---------------------------|----------| | PostgreSQL | Ecto.Adapters.Postgres | Postgrex | | SQLite | Ecto.Adapters.SQLite3 | Exqlite | | MySQL | Ecto.Adapters.MyXQL | MyXQL | | MariaDB | Ecto.Adapters.MyXQL | MyXQL | | SQL Server | Ecto.Adapters.Tds | Tds | | ClickHouse | Ecto.Adapters.ClickHouse | Ch | ## Configuration # config/config.exs config :my_app, SqlKit, root_sql_dir: "priv/repo/sql" # default # config/dev.exs and config/test.exs config :my_app, SqlKit, load_sql: :dynamic # read from disk at runtime # config/prod.exs (or rely on default) config :my_app, SqlKit, load_sql: :compiled # use compile-time embedded SQL ## Options - `:otp_app` (required) - Your application name - `:repo` - The Ecto repo module to use for queries (required unless `:backend` is specified) - `:backend` - Alternative to `:repo` for non-Ecto databases. Currently supports: - `{:duckdb, pool: PoolName}` - Use a DuckDB connection pool - `:dirname` (required) - Subdirectory within root_sql_dir for this module's SQL files - `:files` (required) - List of SQL filenames to load Note: You must specify either `:repo` or `:backend`, but not both. """ alias SqlKit.Config alias SqlKit.Helpers # ============================================================================ # Standalone Query Functions # ============================================================================ @typedoc """ A backend for executing SQL queries. Can be: - An Ecto repo module (e.g., `MyApp.Repo`) - A `SqlKit.DuckDB.Connection` struct (for direct DuckDB connections) - A `SqlKit.DuckDB.Pool` name (atom) for pooled DuckDB connections """ @type backend :: Ecto.Repo.t() | struct() | atom() @doc """ Executes a SQL query and returns all rows as a list of maps or structs. Raises on query execution errors. This matches [`Ecto.Repo.all/2`](https://hexdocs.pm/ecto/Ecto.Repo.html#c:all/2) semantics. ## Backend The first argument can be: - An Ecto repo module (e.g., `MyApp.Repo`) - A `SqlKit.DuckDB.Connection` struct - A `SqlKit.DuckDB.Pool` name (atom) ## Options - `:as` - Struct module to cast results into - `:unsafe_atoms` - If `true`, uses `String.to_atom/1` instead of `String.to_existing_atom/1` for column names. Default: `false` ## Examples SqlKit.query_all(MyApp.Repo, "SELECT * FROM users") # => [%{id: 1, name: "Alice"}, %{id: 2, name: "Bob"}] SqlKit.query_all(MyApp.Repo, "SELECT * FROM users WHERE age > $1", [21]) # => [%{id: 1, name: "Alice", age: 30}] SqlKit.query_all(MyApp.Repo, "SELECT * FROM users", [], as: User) # => [%User{id: 1, name: "Alice"}, %User{id: 2, name: "Bob"}] # With DuckDB connection {:ok, conn} = SqlKit.DuckDB.connect(":memory:") SqlKit.query_all(conn, "SELECT 1 as num", []) # => [%{num: 1}] # With DuckDB pool SqlKit.query_all(MyApp.DuckDBPool, "SELECT * FROM events", []) """ @spec query_all(backend(), String.t(), list() | map(), keyword()) :: [map() | struct()] def query_all(backend, sql, params \\ [], opts \\ []) do SqlKit.Query.all(backend, sql, params, opts) end @doc """ Executes a SQL query and returns exactly one row as a map or struct. Raises `SqlKit.NoResultsError` if no rows are returned. Raises `SqlKit.MultipleResultsError` if more than one row is returned. See `query_all/4` for backend documentation. ## Options - `:as` - Struct module to cast result into - `:unsafe_atoms` - If `true`, uses `String.to_atom/1` instead of `String.to_existing_atom/1` for column names. Default: `false` - `:query_name` - Custom identifier for exceptions (defaults to truncated SQL) ## Examples SqlKit.query_one!(MyApp.Repo, "SELECT * FROM users WHERE id = $1", [1]) # => %{id: 1, name: "Alice"} SqlKit.query_one!(MyApp.Repo, "SELECT * FROM users WHERE id = $1", [1], as: User) # => %User{id: 1, name: "Alice"} """ @spec query_one!(backend(), String.t(), list() | map(), keyword()) :: map() | struct() def query_one!(backend, sql, params \\ [], opts \\ []) do SqlKit.Query.one!(backend, sql, params, opts) end @doc """ Executes a SQL query and returns one row as a map or struct, or nil if no results. Returns the result directly, or nil on no results. Raises `SqlKit.MultipleResultsError` if more than one row is returned. Raises on query execution errors. This matches [`Ecto.Repo.one/2`](https://hexdocs.pm/ecto/Ecto.Repo.html#c:one/2) semantics. See `query_all/4` for backend documentation. ## Options - `:as` - Struct module to cast result into - `:unsafe_atoms` - If `true`, uses `String.to_atom/1` instead of `String.to_existing_atom/1` for column names. Default: `false` - `:query_name` - Custom identifier for exceptions (defaults to truncated SQL) ## Examples SqlKit.query_one(MyApp.Repo, "SELECT * FROM users WHERE id = $1", [1]) # => %{id: 1, name: "Alice"} SqlKit.query_one(MyApp.Repo, "SELECT * FROM users WHERE id = $1", [999]) # => nil """ @spec query_one(backend(), String.t(), list() | map(), keyword()) :: map() | struct() | nil def query_one(backend, sql, params \\ [], opts \\ []) do SqlKit.Query.one(backend, sql, params, opts) end @doc """ Alias for `query_one!/4`. See `query_one!/4` documentation. """ @spec query!(backend(), String.t(), list() | map(), keyword()) :: map() | struct() def query!(backend, sql, params \\ [], opts \\ []) do SqlKit.Query.one!(backend, sql, params, opts) end @doc """ Alias for `query_one/4`. See `query_one/4` documentation. """ @spec query(backend(), String.t(), list() | map(), keyword()) :: map() | struct() | nil def query(backend, sql, params \\ [], opts \\ []) do SqlKit.Query.one(backend, sql, params, opts) end # ============================================================================ # File-Based Macro # ============================================================================ # Validates backend configuration at compile time. # Returns {:ecto, repo_module} or {:duckdb, %{pool: pool_name}} @doc false def validate_backend_config!(opts, caller) do repo = Keyword.get(opts, :repo) backend = Keyword.get(opts, :backend) cond do repo != nil and backend != nil -> raise CompileError, description: "Cannot specify both :repo and :backend options. Use :repo for Ecto repos or :backend for DuckDB pools.", file: caller.file, line: caller.line repo != nil -> expanded_repo = Macro.expand(repo, caller) if not is_atom(expanded_repo) do raise CompileError, description: ":repo must be an atom (module name). Got: #{inspect(repo)}", file: caller.file, line: caller.line end {:ecto, expanded_repo} backend != nil -> validate_backend_option!(backend, caller) true -> raise CompileError, description: "Missing required option: either :repo or :backend must be specified.", file: caller.file, line: caller.line end end defp validate_backend_option!({:duckdb, duckdb_opts}, caller) when is_list(duckdb_opts) do pool = Keyword.get(duckdb_opts, :pool) if pool == nil do raise CompileError, description: "DuckDB backend requires :pool option. Example: backend: {:duckdb, pool: MyApp.DuckDBPool}", file: caller.file, line: caller.line end expanded_pool = Macro.expand(pool, caller) if not is_atom(expanded_pool) do raise CompileError, description: "DuckDB :pool must be an atom (module name). Got: #{inspect(pool)}", file: caller.file, line: caller.line end {:duckdb, %{pool: expanded_pool}} end defp validate_backend_option!(other, caller) do raise CompileError, description: "Invalid :backend option. Expected {:duckdb, pool: PoolName}, got: #{inspect(other)}", file: caller.file, line: caller.line end defmacro __using__(opts) do otp_app = Keyword.fetch!(opts, :otp_app) dirname = Keyword.fetch!(opts, :dirname) files = Keyword.fetch!(opts, :files) # Validate backend configuration - either :repo or :backend required {backend_type, backend_config} = validate_backend_config!(opts, __CALLER__) # Build the SQL directory path at compile time root_sql_dir = Config.root_sql_dir(otp_app) sql_dir = Path.join(root_sql_dir, dirname) # Get the calling module for attribute registration calling_module = __CALLER__.module # Register and embed each SQL file as a module attribute for filename <- files do file_path = Path.join(sql_dir, filename) if File.exists?(file_path) do sql_atom = Helpers.file_atom(filename) sql_content = File.read!(file_path) Module.register_attribute(calling_module, sql_atom, persist: true) Module.put_attribute(calling_module, sql_atom, sql_content) else raise CompileError, description: "SQL file not found: #{filename} at #{file_path}", file: __CALLER__.file, line: __CALLER__.line end end quote do @otp_app unquote(otp_app) @backend_type unquote(backend_type) @backend_config unquote(Macro.escape(backend_config)) @sql_dir unquote(sql_dir) @doc """ Loads a SQL string from a file. In `:compiled` mode (default/production), returns the SQL embedded at compile time. In `:dynamic` mode (dev/test), reads the file from disk for latest changes. Returns `{:ok, sql}` on success, `{:error, reason}` on failure. """ @spec load(filename :: String.t()) :: {:ok, String.t()} | {:error, term()} def load(filename) do {:ok, load!(filename)} rescue e -> {:error, e} end @doc """ Loads a SQL string from a file. In `:compiled` mode (default/production), returns the SQL embedded at compile time. In `:dynamic` mode (dev/test), reads the file from disk for latest changes. Raises on error. """ # sobelow_skip ["Traversal.FileModule"] @spec load!(filename :: String.t()) :: String.t() def load!(filename) do sql_atom = SqlKit.Helpers.file_atom(filename) # Verify the file was registered at compile time sql_content = __MODULE__.__info__(:attributes)[sql_atom] if sql_content == nil do raise "SQL file '#{filename}' was not included in the :files list for #{inspect(__MODULE__)}" end case SqlKit.Config.load_sql(@otp_app) do :compiled -> sql_content :dynamic -> path = Path.join(@sql_dir, filename) File.read!(path) end end @doc """ Executes a SQL query and returns all rows as a list of maps or structs. Raises on query execution errors. This matches [`Ecto.Repo.all/2`](https://hexdocs.pm/ecto/Ecto.Repo.html#c:all/2) semantics. ## Options - `:as` - Struct module to cast results into - `:unsafe_atoms` - If `true`, uses `String.to_atom/1` instead of `String.to_existing_atom/1` for column names. Default: `false` ## Examples SQL.query_all("users.sql", [company_id]) # => [%{id: 1, name: "Alice"}, %{id: 2, name: "Bob"}] SQL.query_all("users.sql", [company_id], as: User) # => [%User{id: 1, name: "Alice"}, %User{id: 2, name: "Bob"}] # ClickHouse uses named parameters as a map ClickHouseSQL.query_all("users.sql", %{company_id: 123}) # => [%{id: 1, name: "Alice"}, %{id: 2, name: "Bob"}] """ @spec query_all(String.t(), list() | map(), keyword()) :: [map() | struct()] def query_all(filename, params \\ [], opts \\ []) do sql = load!(filename) backend = get_backend() SqlKit.Query.all(backend, sql, params, opts) end # Returns the configured backend for query execution. # For Ecto repos, returns the repo module. # For DuckDB pools, returns a pool reference struct. @doc false case @backend_type do :ecto -> def get_backend, do: @backend_config :duckdb -> @pool_name @backend_config.pool def get_backend, do: SqlKit.DuckDB.Pool.pool(@pool_name) end @doc """ Executes a SQL query and returns a single row as a map or struct, or nil if no results. Returns the result directly, or nil on no results. Raises `SqlKit.MultipleResultsError` if more than one row is returned. Raises on query execution errors. This matches [`Ecto.Repo.one/2`](https://hexdocs.pm/ecto/Ecto.Repo.html#c:one/2) semantics. ## Options - `:as` - Struct module to cast result into - `:unsafe_atoms` - If `true`, uses `String.to_atom/1` instead of `String.to_existing_atom/1` for column names. Default: `false` ## Examples SQL.query_one("user.sql", [user_id]) # => %{id: 1, name: "Alice"} SQL.query_one("missing.sql", [999]) # => nil # ClickHouse uses named parameters as a map ClickHouseSQL.query_one("user.sql", %{user_id: 1}) # => %{id: 1, name: "Alice"} """ @spec query_one(String.t(), list() | map(), keyword()) :: map() | struct() | nil def query_one(filename, params \\ [], opts \\ []) do case query_all(filename, params, opts) do [] -> nil [row] -> row rows -> raise SqlKit.MultipleResultsError, filename: filename, count: length(rows) end end @doc """ Executes a SQL query and returns a single row as a map or struct. Raises `SqlKit.NoResultsError` if no rows are returned. Raises `SqlKit.MultipleResultsError` if more than one row is returned. This matches [`Ecto.Repo.one!/2`](https://hexdocs.pm/ecto/Ecto.Repo.html#c:one!/2) semantics. ## Options - `:as` - Struct module to cast result into - `:unsafe_atoms` - If `true`, uses `String.to_atom/1` instead of `String.to_existing_atom/1` for column names. Default: `false` ## Examples SQL.query_one!("user.sql", [user_id]) # => %{id: 1, name: "Alice"} SQL.query_one!("user.sql", [user_id], as: User) # => %User{id: 1, name: "Alice"} # ClickHouse uses named parameters as a map ClickHouseSQL.query_one!("user.sql", %{user_id: 1}) # => %{id: 1, name: "Alice"} """ @spec query_one!(String.t(), list() | map(), keyword()) :: map() | struct() def query_one!(filename, params \\ [], opts \\ []) do case query_all(filename, params, opts) do [] -> raise SqlKit.NoResultsError, filename: filename [row] -> row rows -> raise SqlKit.MultipleResultsError, filename: filename, count: length(rows) end end @doc """ Alias for `query_one/3`. See `query_one/3` documentation. """ @spec query(String.t(), list() | map(), keyword()) :: map() | struct() | nil defdelegate query(filename, params \\ [], opts \\ []), to: __MODULE__, as: :query_one @doc """ Alias for `query_one!/3`. See `query_one!/3` documentation. """ @spec query!(String.t(), list() | map(), keyword()) :: map() | struct() defdelegate query!(filename, params \\ [], opts \\ []), to: __MODULE__, as: :query_one! # DuckDB-specific streaming functions # Only defined when backend is :duckdb if @backend_type == :duckdb do @doc """ Executes a SQL query and streams results through a callback function. Only available for DuckDB backends. The connection is held for the duration of the callback, which receives a stream of result chunks. ## Examples # Count rows without loading all into memory MyApp.Analytics.SQL.with_stream!("large_query.sql", [], fn stream -> stream |> Stream.flat_map(& &1) |> Enum.reduce(0, fn _row, count -> count + 1 end) end) # Process first 100 rows MyApp.Analytics.SQL.with_stream!("events.sql", [~D[2024-01-01]], fn stream -> stream |> Stream.flat_map(& &1) |> Enum.take(100) end) """ @spec with_stream!(String.t(), list(), (Enumerable.t() -> result)) :: result when result: term() def with_stream!(filename, params \\ [], fun) do sql = load!(filename) pool = get_backend() SqlKit.DuckDB.Pool.with_stream!(pool, sql, params, fun) end @doc """ Like `with_stream!/3` but also provides column names to the callback. The callback receives `{columns, stream}` where `columns` is a list of column names and `stream` is the chunk stream. ## Examples MyApp.Analytics.SQL.with_stream_and_columns!("users.sql", [], fn {cols, stream} -> IO.inspect(cols) # => ["id", "name", "age"] stream |> Stream.flat_map(& &1) |> Enum.to_list() end) """ @spec with_stream_and_columns!( String.t(), list(), ({[String.t()], Enumerable.t()} -> result) ) :: result when result: term() def with_stream_and_columns!(filename, params \\ [], fun) do sql = load!(filename) pool = get_backend() SqlKit.DuckDB.Pool.with_stream_and_columns!(pool, sql, params, fun) end end end end # ============================================================================ # Utility Functions # ============================================================================ @doc false # Extracts columns and rows from an Ecto database driver result. @spec extract_result(struct() | map()) :: {[String.t()], [list()]} def extract_result(%{columns: columns, rows: rows}), do: {columns, rows} def extract_result(%{__struct__: struct}) do raise ArgumentError, "Unsupported query result type: #{inspect(struct)}." end def extract_result(other) do raise ArgumentError, "Unsupported query result type: #{inspect(other)}. " end @doc false # Transforms raw query result columns and rows into a list of maps or structs. # sobelow_skip ["DOS.StringToAtom"] @spec transform_rows([String.t()], [list()], keyword()) :: [map() | struct()] def transform_rows(columns, rows, opts \\ []) do unsafe_atoms = Keyword.get(opts, :unsafe_atoms, false) atom_columns = if unsafe_atoms do Enum.map(columns, &String.to_atom/1) else Enum.map(columns, &String.to_existing_atom/1) end struct_mod = Keyword.get(opts, :as) Enum.map(rows, fn row -> map = atom_columns |> Enum.zip(row) |> Map.new() if struct_mod do struct!(struct_mod, map) else map end end) end end