neosqlite.gridfs.gridfs_bucket module

class neosqlite.gridfs.gridfs_bucket.GridFSBucket(db: Connection, bucket_name: str = 'fs', chunk_size_bytes: int = 261120, write_concern: dict[str, Any] | None = None, read_preference: Any | None = None, disable_md5: bool = False)

Bases: object

A GridFSBucket-like class for storing large files in SQLite.

This implementation provides a PyMongo-compatible interface for GridFS functionality using SQLite as the backend storage.

__init__(db: Connection, bucket_name: str = 'fs', chunk_size_bytes: int = 261120, write_concern: dict[str, Any] | None = None, read_preference: Any | None = None, disable_md5: bool = False)

Initialize a new GridFSBucket instance.

Parameters:

• db – SQLite database connection

• bucket_name – The bucket name for the GridFS files (default: “fs”)

• chunk_size_bytes – The chunk size in bytes (default: 255KB)

• write_concern – Write concern settings (simulated for compatibility)

• read_preference – Read preference settings (not applicable to SQLite)

• disable_md5 – Disable MD5 checksum calculation for performance

_migrate_legacy_tables_if_needed()

Migrate legacy GridFS tables from dot-based names to underscore-based names.

_apply_write_concern()

Apply write concern settings to SQLite connection.

_create_collections()

Create the files and chunks collections (tables) if they don’t exist.

_migrate_table_schema()

Migrate existing tables to add new columns for content_type and aliases.

_serialize_metadata(metadata: dict[str, Any] | None) → str | None

Serialize metadata to JSON string.

Parameters:

metadata – Metadata dictionary to serialize

Returns:

JSON string representation or None

_deserialize_metadata(metadata_str: str | None) → dict[str, Any] | None

Deserialize metadata from JSON string.

Parameters:

metadata_str – JSON string representation of metadata

Returns:

Metadata dictionary or None

_force_sync_if_needed()

Force database synchronization if write concern requires it.

upload_from_stream(filename: str, source: bytes | IOBase, chunk_size_bytes: int | None = None, metadata: dict[str, Any] | None = None, session: Any | None = None) → ObjectId

Uploads a user file to a GridFS bucket.

Reads the contents of the user file from source and uploads it as chunks in the chunks collection. After all the chunks have been uploaded, it creates a file document in the files collection.

Parameters:

• filename – The name of the file to upload

• source – The source data (bytes or file-like object)

• chunk_size_bytes – Bytes per chunk (defaults to bucket’s chunk_size_bytes)

• metadata – Optional metadata for the file

• session – A ClientSession for transactions.

Returns:

The ObjectId of the uploaded file document

_insert_chunks(file_id: int, data: bytes)

Split data into chunks and insert them into the chunks collection.

Parameters:

• file_id – The ID of the file document

• data – The data to be chunked

download_to_stream(file_id: ObjectId | str | int, destination: IOBase) → None

Downloads the contents of the stored file specified by file_id and writes the contents to destination.

Parameters:

• file_id – The _id of the file document (ObjectId, hex string, or integer ID)

• destination – A file-like object to which the file contents will be written

_get_integer_id_for_file(file_id: ObjectId | str | int) → int | None

Convert a file identifier (ObjectId, hex string, or integer) to an integer ID.

Parameters:

file_id – The file identifier (ObjectId, hex string, or integer ID)

Returns:

The integer ID corresponding to the file, or None if not found

download_to_stream_by_name(filename: str, destination: IOBase, revision: int = -1) → None

Downloads the contents of the stored file specified by filename and writes the contents to destination.

Parameters:

• filename – The name of the file to download

• destination – A file-like object to which the file contents will be written

• revision – The revision of the file to download (default: -1 for latest)

_get_file_id_by_name(filename: str, revision: int = -1) → int

Get the file ID for a given filename and revision.

Parameters:

• filename – The name of the file

• revision – The revision number (-1 for latest, 0 for first, etc.)

Returns:

The integer _id of the file document

open_download_stream(file_id: ObjectId | str | int) → GridOut

Opens a stream to read the contents of the stored file specified by file_id.

Parameters:

file_id – The _id of the file document (ObjectId, hex string, or integer ID)

Returns:

A GridOut instance to read the file contents

open_download_stream_by_name(filename: str, revision: int = -1) → GridOut

Opens a stream to read the contents of the stored file specified by filename.

Parameters:

• filename – The name of the file to read

• revision – The revision of the file to read (default: -1 for latest)

Returns:

A GridOut instance to read the file contents

delete(file_id: ObjectId | str | int) → None

Given a file_id, delete the stored file’s files collection document and associated chunks from a GridFS bucket.

Parameters:

file_id – The _id of the file document (ObjectId, hex string, or integer ID)

find(filter: dict[str, Any] | None = None, session: Any | None = None) → GridOutCursor

Find and return the files collection documents that match filter.

Parameters:

• filter – The filter to apply when searching for files

• session – A ClientSession for transactions.

Returns:

A GridOutCursor instance

open_upload_stream(filename: str, chunk_size_bytes: int | None = None, metadata: dict[str, Any] | None = None, session: Any | None = None) → GridIn

Opens a stream for writing a file to a GridFS bucket.

Parameters:

• filename – The name of the file to upload

• chunk_size_bytes – Bytes per chunk (defaults to bucket’s chunk_size_bytes)

• metadata – Optional metadata for the file

• session – A ClientSession for transactions.

Returns:

A GridIn instance to write the file contents

upload_from_stream_with_id(file_id: ObjectId | int, filename: str, source: bytes | IOBase, chunk_size_bytes: int | None = None, metadata: dict[str, Any] | None = None, session: Any | None = None)

Uploads a user file to a GridFS bucket with a custom file id.

Parameters:

• file_id – The custom _id for the file document (ObjectId or integer ID)

• filename – The name of the file to upload

• source – The source data (bytes or file-like object)

• chunk_size_bytes – Bytes per chunk (defaults to bucket’s chunk_size_bytes)

• metadata – Optional metadata for the file

• session – A ClientSession for transactions.

open_upload_stream_with_id(file_id: ObjectId | int, filename: str, metadata: dict[str, Any] | None = None, content_type: str | None = None, aliases: list[str] | None = None) → GridIn

Opens a stream for writing a file to a GridFS bucket with a custom file id.

Parameters:

• file_id – The custom _id for the file document (ObjectId or integer ID)

• filename – The name of the file to upload

• metadata – Optional metadata for the file

• content_type – Optional MIME type of the file

• aliases – Optional list of alternative names for the file

Returns:

A GridIn instance to write the file contents

delete_by_name(filename: str) → None

Delete all stored file documents and associated chunks with the given filename.

Parameters:

filename – The name of the file to delete

rename(file_id: ObjectId | str | int, new_filename: str) → None

Rename a stored file with the specified file_id to a new filename.

Parameters:

• file_id – The _id of the file to rename (ObjectId, hex string, or integer ID)

• new_filename – The new name for the file

rename_by_name(filename: str, new_filename: str) → None

Rename all stored files with the specified filename to a new filename.

Parameters:

• filename – The current name of the files to rename

• new_filename – The new name for the files

drop() → None

Remove all files and chunks from the bucket.

This method deletes all data in the GridFS bucket, including all files and their associated chunks.

list() → list[str]

List all unique filenames in the GridFS bucket.

Returns:

A list of unique filenames