neosqlite.gridfs package

Submodules

• neosqlite.gridfs.errors module
  • NeoSQLiteError
    • NeoSQLiteError.__init__()
    • NeoSQLiteError.has_error_label()
    • NeoSQLiteError._add_error_label()
    • NeoSQLiteError._remove_error_label()
    • NeoSQLiteError.timeout
  • GridFSError
  • NoFile
  • FileExists
  • CorruptGridFile
  • PyMongoError
• neosqlite.gridfs.grid_file module
  • GridIn
    • GridIn.__init__()
    • GridIn._serialize_aliases()
    • GridIn._serialize_metadata()
    • GridIn._deserialize_metadata()
    • GridIn._deserialize_aliases()
    • GridIn._force_sync_if_needed()
    • GridIn.write()
    • GridIn._flush_chunk()
    • GridIn._create_file_document()
    • GridIn._get_file_id()
    • GridIn.close()
  • GridOut
    • GridOut.__init__()
    • GridOut._id
    • GridOut._file_id
    • GridOut._deserialize_metadata()
    • GridOut._deserialize_aliases()
    • GridOut.read()
    • GridOut._load_chunk()
    • GridOut.filename
    • GridOut.length
    • GridOut.chunk_size
    • GridOut.upload_date
    • GridOut.md5
    • GridOut.metadata
    • GridOut.content_type
    • GridOut.aliases
    • GridOut.close()
  • GridOutCursor
    • GridOutCursor.__init__()
• neosqlite.gridfs.gridfs_bucket module
  • GridFSBucket
    • GridFSBucket.__init__()
    • GridFSBucket._migrate_legacy_tables_if_needed()
    • GridFSBucket._apply_write_concern()
    • GridFSBucket._create_collections()
    • GridFSBucket._migrate_table_schema()
    • GridFSBucket._serialize_metadata()
    • GridFSBucket._deserialize_metadata()
    • GridFSBucket._force_sync_if_needed()
    • GridFSBucket.upload_from_stream()
    • GridFSBucket._insert_chunks()
    • GridFSBucket.download_to_stream()
    • GridFSBucket._get_integer_id_for_file()
    • GridFSBucket.download_to_stream_by_name()
    • GridFSBucket._get_file_id_by_name()
    • GridFSBucket.open_download_stream()
    • GridFSBucket.open_download_stream_by_name()
    • GridFSBucket.delete()
    • GridFSBucket.find()
    • GridFSBucket.open_upload_stream()
    • GridFSBucket.upload_from_stream_with_id()
    • GridFSBucket.open_upload_stream_with_id()
    • GridFSBucket.delete_by_name()
    • GridFSBucket.rename()
    • GridFSBucket.rename_by_name()
    • GridFSBucket.drop()
    • GridFSBucket.list()
• neosqlite.gridfs.gridfs_legacy module
  • GridFS
    • GridFS.__init__()
    • GridFS.put()
    • GridFS.get()
    • GridFS.get_version()
    • GridFS.get_last_version()
    • GridFS.delete()
    • GridFS.list()
    • GridFS.find()
    • GridFS.find_one()
    • GridFS.exists()
    • GridFS.new_file()
• neosqlite.gridfs.utils module
  • serialize_metadata()
  • deserialize_metadata()
  • serialize_aliases()
  • deserialize_aliases()
  • force_sync_if_needed()

Module contents

exception neosqlite.gridfs.CorruptGridFile(message: str = '', error_labels=None)

Bases: GridFSError

Raised when a file in GridFS is corrupt or incomplete.

exception neosqlite.gridfs.FileExists(message: str = '', error_labels=None)

Bases: GridFSError

Raised when trying to create a file that already exists in GridFS.

class neosqlite.gridfs.GridFS(db: Connection, collection_name: str = 'fs')

Bases: object

A legacy GridFS interface for storing and retrieving files in SQLite.

This class provides the legacy PyMongo-compatible GridFS interface, which is simpler to use than the GridFSBucket API but less flexible.

__init__(db: Connection, collection_name: str = 'fs')

Initialize a new GridFS instance.

Parameters:

• db – SQLite database connection

• collection_name – The collection name for the GridFS files (default: “fs”)

put(data: bytes | IOBase, filename: str | None = None, **kwargs: Any) → ObjectId

Put data into GridFS.

Parameters:

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

• filename – The filename to use (optional)

• **kwargs – Additional metadata fields

Returns:

The ObjectId of the stored file document

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

Get a file from GridFS by its _id.

Parameters:

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

Returns:

A GridOut instance for reading the file

get_version(filename: str, version: int = -1) → GridOut

Get a file from GridFS by filename and version.

Parameters:

• filename – The name of the file to retrieve

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

Returns:

A GridOut instance for reading the file

get_last_version(filename: str) → GridOut

Get the most recent version of a file from GridFS by filename.

Parameters:

filename – The name of the file to retrieve

Returns:

A GridOut instance for reading the file

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

Delete a file from GridFS by its _id.

Parameters:

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

list() → list

List all filenames in GridFS.

Returns:

A list of filenames

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

Find files in GridFS that match the filter.

Parameters:

filter – The filter to apply when searching for files

Returns:

A GridOutCursor instance

find_one(filter: dict[str, Any] | None = None) → GridOut | None

Find a single file in GridFS that matches the filter.

Parameters:

filter – The filter to apply when searching for files

Returns:

A GridOut instance for reading the file, or None if not found

exists(file_id: ObjectId | str | int | None = None, **kwargs: Any) → bool

Check if a file exists in GridFS.

Parameters:

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

• **kwargs – Additional filter criteria (e.g., filename=”test.txt”)

Returns:

True if the file exists, False otherwise

new_file(**kwargs: Any) → GridIn

Create a new file in GridFS and return a GridIn instance to which data can be written.

Parameters:

**kwargs – Arguments to pass to the GridIn constructor (e.g., filename, metadata)

Returns:

A GridIn instance for writing the file contents

class neosqlite.gridfs.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

exception neosqlite.gridfs.NoFile(message: str = '', error_labels=None)

Bases: GridFSError

Raised when trying to access a non-existent file in GridFS.