Implementation:Iterative Dvc Repo Imp Db

Knowledge Sources	Iterative_Dvc
Domains	Data_Import, Database
Last Updated	2026-02-10 10:00 GMT

Overview

The Repo_Imp_Db implementation imports data from a database query or table into a DVC-tracked stage. It resides in dvc/repo/imp_db.py (64 lines) and is the core logic behind the dvc import-db command.

from dvc.repo.imp_db import imp_db

Function Signature

@locked
@scm_context
def imp_db(
    self: "Repo",
    sql: Optional[str] = None,
    table: Optional[str] = None,
    frozen: bool = True,
    output_format: str = "csv",
    out: Optional[str] = None,
    force: bool = False,
    connection: Optional[str] = None,
):

Parameters

Parameter	Type	Default	Description
self	Repo	N/A	The DVC repository instance
sql	Optional[str]	None	SQL query to execute; mutually required with table (one must be set)
table	Optional[str]	None	Database table name to import; mutually required with sql (one must be set)
frozen	bool	True	Whether the resulting stage should be frozen (not re-executed on dvc repro)
output_format	str	"csv"	Output file format; must be "csv" or "json"
out	Optional[str]	None	Output file path; defaults to <table_or_results>.<format>
force	bool	False	Overwrite the output file if it already exists
connection	Optional[str]	None	Named database connection from the DVC configuration

Return Value

Returns the created Stage object representing the database import pipeline stage.

Internal Mechanics

Input Validation

The function asserts that either sql or table is provided, and that output_format is one of the supported formats:

assert sql or table
assert output_format in ("csv", "json")

Database Configuration

A database configuration dictionary is constructed using funcy.compact to exclude None values:

from funcy import compact

db: dict[str, str] = compact(
    {
        "connection": connection,
        "file_format": output_format,
        "query": sql,
        "table": table,
    }
)

Output Path Resolution

The output file name defaults to <table_name>.<format> if a table is specified, or results.<format> if a SQL query is used instead:

file_name = table or "results"
out = out or f"{file_name}.{output_format}"
out = resolve_output(".", out, force=force)

Stage Creation

A single-stage pipeline is created via self.stage.create with the database configuration passed as the db parameter:

path, wdir, out = resolve_paths(self, out, always_local=True)
stage = self.stage.create(
    single_stage=True,
    validate=False,
    fname=path,
    deps=[None],
    wdir=wdir,
    outs=[out],
    db=db,
)

Graph Validation

Before running, the function checks the pipeline graph for output duplication:

try:
    self.check_graph(stages={stage})
except OutputDuplicationError as exc:
    raise OutputDuplicationError(exc.output, set(exc.stages) - {stage})

Execution and Persistence

The stage is executed immediately with stage.run(), then frozen (if frozen=True) and saved to disk with stage.dump().

Usage Example

from dvc.repo import Repo

with Repo() as repo:
    # Import a table as CSV
    stage = repo.imp_db(table="users", connection="mydb")

    # Import via SQL query as JSON
    stage = repo.imp_db(
        sql="SELECT * FROM orders WHERE date > '2025-01-01'",
        output_format="json",
        out="recent_orders.json",
        connection="production_db",
    )

Decorators

Decorator	Purpose
@locked	Ensures the repository lock is held during execution
@scm_context	Manages SCM (Git) context for tracking generated files

Dependencies

Module	Purpose
funcy.compact	Removes None values from the database configuration dictionary
dvc.exceptions.OutputDuplicationError	Raised when a stage output conflicts with an existing output
dvc.repo.scm_context	Decorator for SCM-aware operations
dvc.utils.resolve_output	Resolves and validates the output file path
dvc.utils.resolve_paths	Resolves path, working directory, and output for stage creation

See Also

• Repo_Imp_Url -- Imports data from URLs instead of databases
• Repo_Get -- Downloads data without creating a pipeline stage