SQLite

The SQLite adapter is a data platform adapter that comes bundled with blackline-core, a powerful data modeling and transformation tool. With the SQLite adapter, you can easily connect to SQLite databases and run SQL commands to extract, transform, and load data. The adapter supports multiple profiles for managing different environments, and you can customize the adapter configuration to suit your specific needs. Whether you're working on a small project or a large-scale data transformation, the SQLite adapter is a reliable and efficient tool that can help you get the job done.

The yaml files are used to instantiate the classes: DataStore, SQLiteDataStore, and SQLiteConnectionConfig

Example cofig

foo_dataset.yml

profiles:
  dev:
    type: sqlite
    config:
      connection:
        database: "dev.db"
        uri: true
  prd:
    type: sqlite
    config:
      connection:
        database: "prd.db"
        uri: true

The connection object is defined by the SQLiteConnectionConfig and excepts any argument that is defined in sqlite3.connect

blackline.models.sqlite.sqlite.SQLiteConnectionConfig

Bases: ConnectionConfig

A data model representing a connection configuration for a SQLite database.

Parameters:

Name	Type	Description	Default
database	str	The name or path to the SQLite database file.	required
timeout	float	The number of seconds to wait before timing out the connection.	5.0
detect_types	int	The type detection flags to use when parsing column types.	0
isolation_level	str	The isolation level to use for the connection.	DEFERRED
check_same_thread	bool	Whether to check if the connection is being used in the same thread it was created in.	True
factory	Any		required
cached_statements	int	The maximum number of prepared statements to cache.	100
uri	bool	Whether to use a URI-style connection string.	False

Source code in BAR /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/models/sqlite/sqlite.py

class SQLiteConnectionConfig(ConnectionConfig):
    """
    A data model representing a connection configuration for a SQLite database.
    """

    database: str = Field(
        ..., description="The name or path to the SQLite database file."
    )
    timeout: float = Field(
        default=5.0,
        description="The number of seconds to wait before timing out the connection.",
    )
    detect_types: int = Field(
        default=0,
        description="The type detection flags to use when parsing column types.",
    )
    isolation_level: str = Field(
        default="DEFERRED",
        description="The isolation level to use for the connection.",
    )
    check_same_thread: bool = Field(
        default=True,
        description="Whether to check if the connection is being used in the same thread it was created in.",  # noqa: E501
    )
    factory: Any = Field(
        default=Connection,
        description="The name of the factory function to use for creating the connection.",  # noqa: E501
    )
    cached_statements: int = Field(
        default=100,
        description="The maximum number of prepared statements to cache.",
    )
    uri: bool = Field(
        default=False, description="Whether to use a URI-style connection string."
    )

    class Config:
        env_prefix = ""