Exports data from database tables for DB Sync module.
Provides functions to fetch records from tables with pagination, handle data serialization, and stream large datasets.
Security Considerations
- Uses parameterized queries to prevent SQL injection
- Refuses a table, schema or
:order_byname outsideSchemaInspector.valid_identifier?/1({:error, :invalid_table_name},{:error, :invalid_column_name}) before any query, on every function here: quoting alone would read a real table named"customer-data" - Refuses the tables that are never synced
(
SchemaInspector.excluded_table?/1,{:error, :table_excluded}) - Validates table names against actual database tables
Example
iex> DataExporter.get_count("users")
{:ok, 150}
iex> DataExporter.fetch_records("users", offset: 0, limit: 100)
{:ok, [
%{"id" => 1, "email" => "user@example.com", ...},
...
]}Summary
Functions
Fetches one page of a table and the position the next page continues from.
Fetches records from a table with pagination.
Gets the exact count of records in a table.
The most records fetch_records/2 returns in one call.
Serializes the value of an array column, a list as Postgrex returned it,
by the type of its elements (ColumnInfo.element_type).
Exports all records from a table as a stream.
Functions
@spec fetch_page(String.t(), keyword()) :: {:ok, %{records: [map()], next_after: %{required(String.t()) => String.t()} | nil}} | {:error, any()}
Fetches one page of a table and the position the next page continues from.
Takes the options of fetch_records/2. With :after (the next_after
of the page before, as PhoenixKitSync.PullCursor.from_params/1 reads
it) the page is the rows after that key and :offset is ignored, so rows
inserted or deleted since the page before neither repeat nor hide a row.
Without it the page is cut by :offset, which is what a peer that does
not send a position gets.
next_after is nil for an empty page, for a key no cursor continues
from and under a custom :order_by; the caller then pages by offset.
{:error, :invalid_cursor} when :after does not fit the table's key.
Fetches records from a table with pagination.
Records are returned as maps with string keys matching column names. All values are JSON-serializable.
Options
:offset- Number of records to skip (default: 0):limit- Maximum records to return (default: 100, max: 1000):schema- Database schema (default: "public"):order_by- Column(s) to order by (default: primary key):after- APhoenixKitSync.PullCursorposition; seefetch_page/2
A table with no primary key (and no :order_by) has no order to page in,
so it is served only when it fits in one page (at most :limit rows,
whatever the :offset): a larger one returns {:error, :no_primary_key}.
@spec get_count(String.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, any()}
Gets the exact count of records in a table.
Options
:schema- Database schema (default: "public")
@spec max_limit() :: pos_integer()
The most records fetch_records/2 returns in one call.
Serializes the value of an array column, a list as Postgrex returned it,
by the type of its elements (ColumnInfo.element_type).
Each element follows the rule for a scalar of that type at every depth of
a multi-dimensional array: a uuid goes as its canonical text, other
non-UTF-8 bytes wrapped in base64, a date or a number as its text or
itself. PhoenixKitSync.DataImporter.prepare_array/2 reads the list back.
@spec stream_records(String.t(), keyword()) :: {:ok, Enumerable.t()} | {:error, any()}
Exports all records from a table as a stream.
Useful for large tables where loading all records into memory
is not practical. Returns a stream that yields batches of records, each
starting after the key the one before ended on (fetch_page/2).
Options
:batch_size- Records per batch (default: 500):schema- Database schema (default: "public")