neosqlite.connection module¶

class neosqlite.connection.Connection(*args: Any, **kwargs: Any)[source]¶

Bases: object

Represents a connection to an NeoSQLite database.

Provides methods for managing collections, executing SQL queries, and handling database lifecycle events. Supports PyMongo-like API.

DEFAULT_TRANSLATION_CACHE_SIZE = 100¶

__init__(*args: Any, **kwargs: Any) → None[source]¶

Initialize a new database connection.

Parameters:

*args – Positional arguments passed to sqlite3.connect().
**kwargs –
Keyword arguments passed to sqlite3.connect(). Special kwargs: - tokenizers: List of tuples (name, path) for FTS5 tokenizers to load - debug: Boolean flag to enable debug printing - name: Optional name for the database (for PyMongo API compatibility) - _is_clone: Internal flag for cloning (not for public use) - journal_mode: Optional journal mode (default: “WAL”) - auto_vacuum: Auto vacuum mode (default: INCREMENTAL).

Can be 0/NONE, 1/FULL, 2/INCREMENTAL, or “NONE”/”FULL”/”INCREMENTAL”. If database has different auto_vacuum setting, migration may be triggered.
- translation_cache: SQL translation cache size (default: 100, 0 to disable)

_collections: dict[str, Collection]¶

_tokenizers: list[tuple[str, str]]¶

debug: bool¶

_translation_cache_size: int | None¶

name: str¶

property db_path: str¶

Get the path to the database file.

Returns:: The database file path.
Return type:: str

connect(*args: Any, **kwargs: Any) → None[source]¶

Establish a connection to the SQLite database.

Configures the database connection with the provided arguments, sets up SQLite-specific settings like isolation level and journal mode, and loads custom FTS5 tokenizers if specified. This method does not return a value.

Parameters:

*args – Positional arguments passed to sqlite3.connect().
**kwargs – Keyword arguments passed to sqlite3.connect().

_register_custom_functions() → None[source]¶: Register custom SQLite functions, including regex operators.

_check_and_migrate_autovacuum(*args: Any, **kwargs: Any) → None[source]¶: Check auto_vacuum setting and migrate if needed. Called after initial connection is established.

_migrate_to_autovacuum(*args: Any, **kwargs: Any) → None[source]¶

Migrate database to a new auto_vacuum mode.

Delegates to migration.migrate_autovacuum() and then reconnects.

cleanup() → None[source]¶: Clean up all collection resources associated with this connection.

close() → None[source]¶

Close the database connection.

Commits any pending transaction and properly closes the underlying SQLite connection. This method ensures resources are released and the connection is no longer usable after being called.

property client: Connection¶

Get the MongoClient instance (returns self for NeoSQLite).

Returns:: The connection instance itself.
Return type:: Connection

property codec_options: Any¶

Get the codec options for this connection.

Returns:: The codec options.
Return type:: Any

property read_preference: Any¶

Get the read preference for this connection.

Returns:: The read preference.
Return type:: Any

property write_concern: Any¶

Get the write concern for this connection.

Returns:: The write concern.
Return type:: Any

property read_concern: Any¶

Get the read concern for this connection.

Returns:: The read concern.
Return type:: Any

start_session(causal_consistency: bool | None = None, default_transaction_options: dict[str, Any] | None = None, **kwargs: Any) → ClientSession[source]¶

Start a new client session for transactions.

This method provides PyMongo-compatible session management by wrapping SQLite’s native ACID transactions.

Parameters:

causal_consistency (bool, optional) – Whether to enable causal consistency. Ignored in NeoSQLite (stored for API compatibility).
default_transaction_options (dict, optional) – Default transaction options.
**kwargs – Additional session arguments.

Returns:

A new ClientSession instance.

Return type:

ClientSession

drop_collection(name: str) → None[source]¶

Drop a collection (table) from the database.

Parameters:: name (str) – The name of the collection (table) to drop. If the table does not exist, the operation is silently ignored due to the use of IF EXISTS in the SQL command.

create_collection(name: str, **kwargs) → Collection[source]¶

Create a new collection with specific options.

Parameters:

name (str) – The name of the collection to create.
**kwargs – Additional options for collection creation.

Returns:

The newly created collection.

Return type:

Collection

Raises:

CollectionInvalid – If a collection with the given name already exists.

get_collection(name: str, **kwargs) → Collection[source]¶

Get a collection by name.

Parameters:

name (str) – The name of the collection to get.
**kwargs – Additional options for collection access.

Returns:

The collection instance.

Return type:

Collection

rename_collection(old_name: str, new_name: str) → None[source]¶

Rename a collection.

Parameters:

old_name (str) – The current name of the collection.
new_name (str) – The new name for the collection.

Raises:

CollectionInvalid – If the old collection doesn’t exist or new name already exists.

list_collection_names() → list[str][source]¶

List all collection names in the database.

Returns:: A list of all collection names in the database, excluding internal SQLite tables (sqlite_sequence, sqlite_stat*, etc.)
Return type:: list[str]

list_collections() → list[dict[str, Any]][source]¶

Get detailed information about collections in the database.

Returns:

A list of dictionaries containing collection information.: Each dictionary has ‘name’ and ‘options’ keys.

Return type:

list[dict[str, Any]]

command(command: str | dict[str, Any], value: Any | None = None, **kwargs: Any) → dict[str, Any][source]¶

Issue a database command and return the response.

This method provides PyMongo-compatible command execution for SQLite. It supports various commands including PRAGMA commands, introspection, and utility commands.

Parameters:

command – The command to execute. Can be: - A string (e.g., “table_info”, “integrity_check”) - A dict with command name as key (e.g., {“ping”: 1})
value – Optional command value (for string commands)
**kwargs – Additional command arguments

Returns:

Command response

Return type:

dict[str, Any]

Supported Commands:

“ping” or {“ping”: 1} - Returns {“ok”: 1}
“serverStatus” - Returns SQLite version info
“listCollections” - Returns collection list
“table_info” - Returns table schema (PRAGMA table_info)
“integrity_check” - Returns integrity check results
“validate” - Returns integrity check for a collection
“reIndex” - Rebuilds indexes for a collection
“foreign_key_check” - Returns foreign key check results
“index_list” - Returns index list for a table
“vacuum” - Runs full VACUUM command
“compact” - MongoDB-compatible compact (see below)
“wal_checkpoint” - WAL checkpoint control (NeoSQLite extension)
“cache_size” - Get/set cache size (NeoSQLite extension)
“busy_timeout” - Get/set busy timeout (NeoSQLite extension)
“query_only” - Get/set read-only mode (NeoSQLite extension)
“analyze” - Runs ANALYZE command

compact Command:

MongoDB-compatible compact implementation with NeoSQLite extensions.

Args:

collection: Collection name (ignored, operates on entire database) dryRun: If true, returns estimate without actually compacting freeSpaceTargetMB: Target in MB (default: 20). Only compacts if

free space >= threshold. Uses incremental vacuum in batches. Set to 0 for full VACUUM instead of incremental.

comment: Optional comment (ignored, for MongoDB compatibility)

Returns:

dryRun: {“estimatedBytesFreed”: <bytes>, “ok”: 1} compact: {“bytesFreed”: <bytes>, “ok”: 1}

Examples:

>>> db.command("compact", "collection")  # 20MB threshold + incremental
>>> db.command("compact", "collection", dryRun=True)  # Estimate only
>>> db.command("compact", "collection", freeSpaceTargetMB=10)  # 10MB threshold + incremental
>>> db.command("compact", "collection", freeSpaceTargetMB=0)  # Full vacuum

Example

>>> db = Connection("test.db")
>>> result = db.command("ping")
>>> print(result)
{'ok': 1.0}

cursor_command(command: str | dict[str, Any], value: Any | None = None, **kwargs: Any) → AggregationCursor[source]¶

Execute a database command and return a cursor for its results.

This method provides PyMongo-compatible cursor-based command execution. It wraps the results of command() in an AggregationCursor, allowing them to be iterated.

Parameters:

command – The command to execute.
value – Optional command value.
**kwargs – Additional command arguments.

Returns:

A cursor over the command results.

Return type:

AggregationCursor

dereference(dbref: dict[str, Any]) → dict[str, Any] | None[source]¶

Resolve a DBRef object.

This method provides PyMongo-compatible DBRef resolution by performing a find_one on the target collection using the provided $id.

Parameters:: dbref – A dictionary representing a DBRef with ‘$ref’ and ‘$id’ keys.
Returns:: The referenced document, or None if not found.
Return type:: dict[str, Any] | None

with_options(codec_options: Any | None = None, read_preference: Any | None = None, write_concern: Any | None = None, read_concern: Any | None = None) → Connection[source]¶

Get a clone of this database with different options.

This method returns a new Connection instance with the specified options. For PyMongo API compatibility, the options are stored but not actively used since SQLite has different semantics.

Parameters:

codec_options (Any, optional) – Codec options for encoding/decoding. Ignored in NeoSQLite (stored for API compatibility).
read_preference (Any, optional) – Read preference for replica sets. Ignored in NeoSQLite (stored for API compatibility).
write_concern (Any, optional) – Write concern for durability settings. Stored for API compatibility.
read_concern (Any, optional) – Read concern for consistency settings. Ignored in NeoSQLite (stored for API compatibility).

Returns:

A new Connection instance with the same underlying database: but with the specified options stored.

Return type:

Connection

Note

NeoSQLite stores these options for PyMongo API compatibility, but they don’t affect SQLite behavior since SQLite doesn’t have replica sets, codec options, or the same consistency/durability model.

Example

>>> db = Connection("test.db")
>>> db_with_options = db.with_options(
...     write_concern={"w": "majority"},
...     read_preference={"mode": "primaryPreferred"}
... )

_apply_write_concern(write_concern: Any) → None[source]¶

Apply write concern to the underlying SQLite connection via PRAGMAs.

w: 0 -> PRAGMA synchronous = OFF
w: 1 -> PRAGMA synchronous = NORMAL
j: True -> PRAGMA synchronous = FULL

Parameters:: write_concern (Any) – The write concern to apply (dict or WriteConcern object).

transaction() → Iterator[None][source]¶

Context manager for handling database transactions.

Ensures atomicity by beginning a transaction on entry, committing on successful exit, and rolling back in case of exceptions. This allows using the connection in a ‘with’ statement to manage transaction boundaries safely.

Yields control to the block, and automatically commits or rolls back based on execution outcome.