feat(sql): load remote SQLite and DuckDB databases

- Load SQLite and DuckDB source files from S3, Azure Blob Storage, and Google Cloud Storage.
- Stage each object for the complete ingestion lifetime with size checks, byte verification, cleanup, and no remote writeback.
- Reuse filesystem credential parsing while keeping credentials out of logs, errors, and object representations.
- Document the remote URI contract and cover parsing, lifecycle, compatibility, and emulator-backed ingestion.

Author: hampsterx

docs/decisions/ADR-001-stage-remote-file-databases.md

@@ -0,0 +1,65 @@
+# ADR-001: Stage remote file databases before SQL ingestion
+
+**Status**: Proposed
+**Date**: 2026-08-03
+
+## Context
+
+SQLite and DuckDB require random access to a database file through their
+existing SQLAlchemy integrations. Object-storage streams do not provide a
+shared database interface that both engines can open directly.
+
+Remote database access needs the same URI and lifecycle semantics across
+supported storage transports. Storage credentials must remain separate from
+logged object locations, and the database file must remain available while dlt
+constructs and consumes its lazy SQL resource.
+
+## Decision
+
+We materialize remote SQLite and DuckDB source files into a run-scoped local
+temporary directory before invoking the existing SQL source path.
+
+Remote sources use a parameter-form URI grammar with an encoded `location`
+value and outer storage options. Initial transport support is limited to
+`s3://`, `az://`, and `gs://`, which share the repository's existing filesystem
+credential parsing and emulator coverage.
+
+The staging operation checks the remote object's reported size and available
+disk space, streams the object while counting bytes, and keeps the local copy
+alive through the complete ingestion run. It removes the copy on normal
+completion and exception exits. The staged database is source-only, and local
+changes are never written back to object storage.
+
+## Alternatives considered
+
+- **Stack storage schemes inside the SQL URI**: rejected because nested URI
+  authorities, query strings, and credential ownership are ambiguous to
+  standard URL parsers.
+- **Use DuckDB `httpfs` or `ATTACH` as the primary path**: rejected because it
+  does not provide the same mechanism for SQLite. Engine-specific fast paths
+  can be added later without changing the public URI grammar.
+- **Expose object storage through an engine-specific VFS**: rejected because
+  there is no shared, maintained random-access interface for both engines and
+  all three launch transports.
+- **Keep a persistent local cache**: rejected because cache invalidation,
+  credential boundaries, and stale-object behavior require a separate policy.
+  Run-scoped staging has explicit ownership and cleanup.
+- **Write staged database changes back remotely**: rejected because concurrent
+  writers and atomic replacement require consistency guarantees that a source
+  ingestion does not need.
+
+## Consequences
+
+- Remote file databases reuse the established SQL reflection, extraction, and
+  loading path.
+- Each run downloads the whole object and requires enough local disk for the
+  database plus any engine sidecar files.
+- Normal success and failure paths remove staged data. A process terminated
+  without cleanup may leave a temporary directory when a persistent staging
+  parent is configured.
+- Credentials remain storage options and are excluded from safe object
+  locations and object representations.
+- Additional transports need an explicit grammar mapping, credential adapter,
+  and integration coverage.
+- A future engine-specific fast path must preserve the same source-only behavior
+  and public URI contract.

docs/index.md

@@ -16,13 +16,14 @@ inherited by [dlt]: `append`, `merge`, and `delete+insert`.
 
 We recommend using [uv](https://github.com/astral-sh/uv) to run `omniload`.
 
-```
+```bash
 pip install uv
 uvx omniload
 ```
 
 Alternatively, if you'd like to install it globally:
-```
+
+```bash
 uv pip install --system omniload
 ```
 
@@ -35,15 +36,15 @@ Check out the {ref}`Quickstart` guide to get started with omniload.
 ### License
 
 The project is licensed under the MIT License, see the [LICENSE] file for details.
-Some components are licensed under the Apache 2.0 license, see the [NOTICE] file for details.
+Some components are licensed under the Apache 2.0 license, see the [NOTICE]
+file for details.
 
 ### Acknowledgements
 
 This project would not have been possible without the amazing work by the
 authors and contributors to [SQLAlchemy], [dlt], and [ingestr], turtles all
 the way down. Kudos.
 
-
 ```{toctree}
 :caption: Commands and adapters
 :maxdepth: 1
@@ -79,13 +80,13 @@ tutorials/*
 :maxdepth: 1
 :hidden:
 :glob:
+decisions/*
 sandbox
 changelog
 contributors
 backlog
 ```
 
-
 [dlt]: https://github.com/dlt-hub/dlt
 [ingestr]: https://bruin-data.github.io/ingestr/
 [LICENSE]: https://github.com/panodata/omniload/blob/main/LICENSE

docs/supported-sources/duckdb.md

@@ -1,16 +1,56 @@
 # DuckDB
+
 DuckDB is an in-memory database designed to be fast and reliable.
 
 omniload supports DuckDB as both a source and destination.
 
 ## URI format
+
 The URI format for DuckDB is as follows:
 
 ```text
 duckdb:///<database-file>
 ```
 
 URI parameters:
+
 - `database-file`: the path to the DuckDB database file
 
-The same URI structure can be used both for sources and destinations. You can read more about SQLAlchemy's DuckDB dialect [here](https://github.com/Mause/duckdb_engine).
+The same URI structure can be used both for sources and destinations. See the
+[DuckDB SQLAlchemy dialect][duckdb-sqlalchemy] for details.
+
+## Remote source files
+
+A DuckDB source file can live in [Amazon S3](s3.md),
+[Azure Blob Storage](azure-storage.md), or
+[Google Cloud Storage](google-cloud-storage.md). Put its percent-encoded object
+URI in the `location` query parameter, then add the storage connector's
+credentials and options as outer query parameters:
+
+```text
+duckdb:///?location=<percent-encoded-s3-az-or-gs-uri>&<storage-options>
+```
+
+For example, this loads `main.events` from `s3://analytics/snapshots/events.duckdb`:
+
+```bash
+LOCATION='s3%3A%2F%2Fanalytics%2Fsnapshots%2Fevents.duckdb'
+CREDENTIALS='access_key_id=ACCESS&secret_access_key=SECRET'
+omniload ingest \
+  --source-uri "duckdb:///?location=${LOCATION}&${CREDENTIALS}" \
+  --source-table 'main.events' \
+  --dest-uri 'duckdb:///local.duckdb' \
+  --dest-table 'raw.events'
+```
+
+Percent-encode every query value that contains reserved characters. This is
+required for credentials containing characters such as `+`, `/`, `=`, `&`, or
+`?`, and for percent-encoded object names. The `location` value itself must not
+contain a query or fragment; put all storage options in the outer query.
+
+omniload downloads the complete object into a run-scoped temporary directory
+before opening it. The file must fit on local disk. The staged copy is removed
+after success or failure, and changes made to it are never written back to
+object storage. Remote file databases are supported as sources only.
+
+[duckdb-sqlalchemy]: https://github.com/Mause/duckdb_engine

docs/supported-sources/sqlite.md

@@ -1,16 +1,57 @@
 # SQLite
-SQLite is a C-language library that implements a small, fast, self-contained, high-reliability, full-featured, SQL database engine.
+
+SQLite is a C-language library that implements a small, fast, self-contained,
+high-reliability, full-featured SQL database engine.
 
 omniload supports SQLite as a source and a destination.
 
 ## URI format
+
 The URI format for SQLite is as follows:
 
 ```text
 sqlite:///<database-file>
 ```
 
 URI parameters:
+
 - `database-file`: the path to the SQLite database file.
 
-The same URI structure can be used both for sources and destinations. You can read more about SQLAlchemy's SQLite dialect [here](https://docs.sqlalchemy.org/en/20/core/engines.html#sqlite).
+The same URI structure can be used both for sources and destinations. See the
+[SQLite SQLAlchemy dialect][sqlite-sqlalchemy] for details.
+
+## Remote source files
+
+An SQLite source file can live in [Amazon S3](s3.md),
+[Azure Blob Storage](azure-storage.md), or
+[Google Cloud Storage](google-cloud-storage.md). Put its percent-encoded object
+URI in the `location` query parameter, then add the storage connector's
+credentials and options as outer query parameters:
+
+```text
+sqlite:///?location=<percent-encoded-s3-az-or-gs-uri>&<storage-options>
+```
+
+For example, this loads `main.events` from `s3://analytics/snapshots/events.sqlite`:
+
+```bash
+LOCATION='s3%3A%2F%2Fanalytics%2Fsnapshots%2Fevents.sqlite'
+CREDENTIALS='access_key_id=ACCESS&secret_access_key=SECRET'
+omniload ingest \
+  --source-uri "sqlite:///?location=${LOCATION}&${CREDENTIALS}" \
+  --source-table 'main.events' \
+  --dest-uri 'duckdb:///local.duckdb' \
+  --dest-table 'raw.events'
+```
+
+Percent-encode every query value that contains reserved characters. This is
+required for credentials containing characters such as `+`, `/`, `=`, `&`, or
+`?`, and for percent-encoded object names. The `location` value itself must not
+contain a query or fragment; put all storage options in the outer query.
+
+omniload downloads the complete object into a run-scoped temporary directory
+before opening it. The file must fit on local disk. The staged copy is removed
+after success or failure, and changes made to it are never written back to
+object storage. Remote file databases are supported as sources only.
+
+[sqlite-sqlalchemy]: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html

pyproject.toml

@@ -98,7 +98,7 @@ dependencies = [
   "google-cloud-bigquery-storage<3",
   "google-cloud-spanner<4",
   "influxdb-client<2",
-  "mq-bridge-py>=0.3.2,<0.4",
+  "mq-bridge-py>=0.3.2,<0.4,!=0.3.9",
   "mysql-connector-python<27",
   "omniload[filesystem]",
   "oracledb<5",

src/dlt_filesystem/source/impl/remote.py

@@ -1,5 +1,3 @@
-import base64
-import json
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Any, Dict, Type
 from urllib.parse import parse_qs, urlparse
@@ -16,7 +14,12 @@
     parse_uri,
     source_selects_single_file,
 )
-from dlt_filesystem.util.auth import AzureBlobAuth, parse_azure_blob_auth
+from dlt_filesystem.util.auth import (
+    azure_blob_filesystem_kwargs,
+    gcs_filesystem_kwargs,
+    parse_azure_blob_auth,
+    s3_filesystem_kwargs,
+)
 
 if TYPE_CHECKING:
     from fsspec import AbstractFileSystem
@@ -52,25 +55,7 @@ def dlt_source(self, uri: str, table: str, **kwargs):
 
         bucket_url = f"gs://{bucket_name}"
 
-        credentials_path = params.pop("credentials_path", [None])[0]
-        credentials_base64 = params.pop("credentials_base64", [None])[0]
-
-        # Merge params into fs kwargs, without overriding kwargs already
-        # supplied by the caller (e.g. filesystem_incremental, column_types).
-        for key, value in params.items():
-            kwargs.setdefault(key, value[0])
-
-        if "token" not in kwargs:
-            credentials = None
-            if credentials_path:
-                credentials = credentials_path
-            elif credentials_base64:
-                credentials = json.loads(base64.b64decode(credentials_base64).decode())
-            else:
-                credentials = "anon"
-            kwargs["token"] = credentials
-
-        fs = self.fs_class(**kwargs)
+        fs = self.fs_class(**gcs_filesystem_kwargs(params, kwargs))
 
         try:
             endpoint: str = determine_endpoint(table, path_to_file)
@@ -133,27 +118,14 @@ def dlt_source(self, uri: str, table: str, **kwargs):
 
         parsed_uri = urlparse(uri)
         source_fields = parse_qs(parsed_uri.query)
-        access_key_id = source_fields.get("access_key_id")
-        if not access_key_id:
-            raise MissingConnectorOption("access_key_id", self.fs_name)
-
-        secret_access_key = source_fields.get("secret_access_key")
-        if not secret_access_key:
-            raise MissingConnectorOption("secret_access_key", self.fs_name)
-
+        fs_kwargs = s3_filesystem_kwargs(source_fields, self.fs_name)
         bucket_name, path_to_file = parse_uri(parsed_uri, table)
         if not bucket_name or not path_to_file:
             raise InvalidBlobTableError(self.fs_name)
 
         bucket_url = f"{self.fs_protocol}://{bucket_name}/"
 
         endpoint_url = source_fields.get("endpoint_url")
-        fs_kwargs: dict = {
-            "key": access_key_id[0],
-            "secret": secret_access_key[0],
-        }
-        if endpoint_url:
-            fs_kwargs["endpoint_url"] = endpoint_url[0]
 
         fs = self.fs_class(**fs_kwargs)
 
@@ -190,35 +162,6 @@ def fs_name(self) -> str:
         return "S3"
 
 
-def _azure_kwargs(auth: AzureBlobAuth):
-    """Return AzureBlobAuth information as dictionary.
-
-    The ingestr-style short names already match adlfs kwargs, so they pass
-    straight through; only the supplied ones are forwarded. ``adlfs`` is
-    imported lazily so the CLI ``--help`` and every non-Azure path never load
-    the Azure SDK (matching the s3fs/gcsfs deferred-import convention).
-    """
-
-    kwargs = {"account_name": auth.account_name}
-    if auth.account_key is not None:
-        kwargs["account_key"] = auth.account_key
-    if auth.sas_token is not None:
-        kwargs["sas_token"] = auth.sas_token
-    if auth.tenant_id is not None:
-        kwargs["tenant_id"] = auth.tenant_id
-    if auth.client_id is not None:
-        kwargs["client_id"] = auth.client_id
-    if auth.client_secret is not None:
-        kwargs["client_secret"] = auth.client_secret
-    if auth.account_host is not None:
-        kwargs["account_host"] = auth.account_host
-    if auth.connection_string is not None:
-        kwargs["connection_string"] = auth.connection_string
-    if auth.api_version is not None:
-        kwargs["api_version"] = auth.api_version
-    return kwargs
-
-
 class AzureSource(FilesystemSource):
     """Azure Blob Storage / ADLS Gen2 source (``az://``, ``adls://``, ``abfss://``).
 
@@ -256,7 +199,7 @@ def dlt_source(self, uri: str, table: str, **kwargs):
 
         bucket_url = f"az://{bucket_name}"
 
-        kwargs.update(_azure_kwargs(auth))
+        kwargs.update(azure_blob_filesystem_kwargs(auth))
         fs = self.fs_class(**kwargs)
 
         try: