MongoDBDestination

class

class MongoDBDestination(
    uri: str,
    collections_with_ids: tuple[str, str | None] | List[tuple[str, str | None]],
    credentials: UserPasswordCredentials = None,
    connection_options: dict = None,
    if_collection_exists: Literal['append', 'replace'] = 'append',
    use_trxs: bool = False,
    docs_per_trx: int = 1000,
    maintain_order: bool = False,
    update_existing: bool = True,
    fail_on_duplicate_key: bool = True,
    log_intermediate_files: bool = False,
    **kwargs,
)

Bases: DestinationPlugin

Categories: destination

Abstract class for output plugins.

Methods: trigger_output(working_dir, *args, **kwargs) Trigger the exporting of the data. This function will receive the resulting data from the dataset function and must store it in the desired location.

Initializes the MongoDBDestination with the configuration desired to store the data.

Parameters

parameter

uristr

The URI of the MongoDB database.

parameter

collections_with_idstuple[str, str] | List[tuple[str, str]]

A tuple or list of tuples with the collection and the name of the field that will be used as the unique identifier. For example, if you want to store the data in a collection called 'my_collection' in database 'my_database' and use the field 'username' as the unique identifier, you would provide the following tuple: ( 'my_database.my_collection', 'username'). If you wanted MongoDB to autogenerate the id, you would provide the following tuple: ('my_database.my_collection', None).

parameter

credentialsUserPasswordCredentials, optional

The credentials to connect with the database. If None, no credentials will be used.

parameter

connection_optionsdict, optional

A dictionary with the options to pass to the pymongo.MongoClient constructor. For example, if you want to set the timeout to 1000 milliseconds, you would provide the following dictionary: {'serverSelectionTimeoutMS': 1000}.

parameter

if_collection_existsLiteral["append", "replace"], optional

The action to take if the collection already exists. If 'append', the data will be appended to the existing collection. If 'replace', the existing collection will be replaced with the new data. Defaults to 'append'.

parameter

use_trxsbool, optional

Whether to use a transaction when storing the data in the database. If True, the data will be stored in a transaction, which will ensure that all the data is stored or none of it is (requires that the database is configured with a replica set). If False, the data will be stored without a transaction, which may lead to inconsistent data in the database. Defaults to False.

parameter

docs_per_trxint, optional

The maximum number of documents to store in a single transaction. If the number of documents to store exceeds this number, the data will be stored in multiple transactions.

parameter

maintain_orderbool, optional

Whether to maintain the order of the documents when storing them in the database. If True, the documents will be stored in the same order as they are in the TableFrame. If False, the documents will be stored in the order that they are processed. Defaults to False.

parameter

update_existingbool, optional

Whether to update the existing documents in the database. If True, the documents will be updated if they already exist in the database. If False, the documents will be inserted without updating the existing documents, and if a document with the same id already exists execution will fail. Defaults to True.

parameter

fail_on_duplicate_keybool, optional

Whether to raise an exception if a document with the same id already exists in the collection. If True, an exception will be raised. If False, the operation will continue without raising an exception. Defaults to True.

parameter

log_intermediate_filesbool, optional

Whether to log when each batch of data is stored in the database. If True, a message will be logged for each batch of data stored. If False, no message will be logged until all the data for a single collection has been stored. Defaults to False.

Properties

property

collections_with_idsList[tuple[str, str | None]]

property

connection_optionsdict

property

credentialsUserPasswordCredentials | None

property

if_collection_existsLiteral['append', 'replace']

property

uristr

Methods

method

chunk

def chunk(working_dir: str, *results: LazyFrame | None) -> List[List[str] | None]

Trigger the exporting of the data to local parquet chunks. This method will receive the resulting data from the user function and must store it in the local system as parquet files, using the working_dir. Note: This method should not materialize the data, it should only store it in the local system.

Parameters:

parameter

working_dirstr

The folder where any files generated must be stored (this refers to temporary files that will be deleted after the execution of the plugin, not the final destination of the data)

parameter

results

The data to be exported. It is a list of polars LazyFrames or None.

Returns:

A list of the intermediate files created

method

stream

def stream(working_dir: str, *results: List[LazyFrame | None] | LazyFrame | NoneType)

Trigger the exporting of the data. This method will receive the resulting data from the user function and must store it in the desired location. Note: this method might materialize the data provided in a single chunk generated by the chunk function if invoked, so chunks should be of an appropriate size.

Parameters:

parameter

working_dirstr

The folder where any intermediate files generated must be stored (this refers to temporary files that will be deleted after the execution of the plugin, not the final destination of the data)

parameter

results

The data to be exported. It is a list of polars LazyFrames or None.

Returns:

None

method

write

def write(files)

This method is used to write the files to the database. It is called from the stream method, and it is not intended to be called directly.