airflow.providers.mongo.hooks.mongo

Hook for Mongo DB

Module Contents

class airflow.providers.mongo.hooks.mongo.MongoHook(conn_id: str = default_conn_name, *args, **kwargs)[source]

Bases: airflow.hooks.base.BaseHook

Interact with Mongo. This hook uses the Mongo conn_id. PyMongo Wrapper to Interact With Mongo Database Mongo Connection Documentation https://docs.mongodb.com/manual/reference/connection-string/index.html You can specify connection string options in extra field of your connection https://docs.mongodb.com/manual/reference/connection-string/index.html#connection-string-options

If you want use DNS seedlist, set srv to True.

ex.

{"srv": true, "replicaSet": "test", "ssl": true, "connectTimeoutMS": 30000}

Parameters

mongo_conn_id -- The Mongo connection id to use when connecting to MongoDB.

conn_name_attr = conn_id[source]
default_conn_name = mongo_default[source]
conn_type = mongo[source]
hook_name = MongoDB[source]
__enter__(self)[source]
__exit__(self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType])[source]
get_conn(self)[source]

Fetches PyMongo Client

close_conn(self)[source]

Closes connection

get_collection(self, mongo_collection: str, mongo_db: Optional[str] = None)[source]

Fetches a mongo collection object for querying.

Uses connection schema as DB unless specified.

aggregate(self, mongo_collection: str, aggregate_query: list, mongo_db: Optional[str] = None, **kwargs)[source]

Runs an aggregation pipeline and returns the results https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.aggregate https://api.mongodb.com/python/current/examples/aggregation.html

find(self, mongo_collection: str, query: dict, find_one: bool = False, mongo_db: Optional[str] = None, **kwargs)[source]

Runs a mongo find query and returns the results https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.find

insert_one(self, mongo_collection: str, doc: dict, mongo_db: Optional[str] = None, **kwargs)[source]

Inserts a single document into a mongo collection https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.insert_one

insert_many(self, mongo_collection: str, docs: dict, mongo_db: Optional[str] = None, **kwargs)[source]

Inserts many docs into a mongo collection. https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.insert_many

update_one(self, mongo_collection: str, filter_doc: dict, update_doc: dict, mongo_db: Optional[str] = None, **kwargs)[source]

Updates a single document in a mongo collection. https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.update_one

Parameters

  • mongo_collection (str) -- The name of the collection to update.

  • filter_doc (dict) -- A query that matches the documents to update.

  • update_doc (dict) -- The modifications to apply.

  • mongo_db (str) -- The name of the database to use. Can be omitted; then the database from the connection string is used.

update_many(self, mongo_collection: str, filter_doc: dict, update_doc: dict, mongo_db: Optional[str] = None, **kwargs)[source]

Updates one or more documents in a mongo collection. https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.update_many

Parameters

  • mongo_collection (str) -- The name of the collection to update.

  • filter_doc (dict) -- A query that matches the documents to update.

  • update_doc (dict) -- The modifications to apply.

  • mongo_db (str) -- The name of the database to use. Can be omitted; then the database from the connection string is used.

replace_one(self, mongo_collection: str, doc: dict, filter_doc: Optional[dict] = None, mongo_db: Optional[str] = None, **kwargs)[source]

Replaces a single document in a mongo collection. https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.replace_one

Note

If no filter_doc is given, it is assumed that the replacement document contain the _id field which is then used as filters.

Parameters

  • mongo_collection (str) -- The name of the collection to update.

  • doc (dict) -- The new document.

  • filter_doc (dict) -- A query that matches the documents to replace. Can be omitted; then the _id field from doc will be used.

  • mongo_db (str) -- The name of the database to use. Can be omitted; then the database from the connection string is used.

replace_many(self, mongo_collection: str, docs: List[dict], filter_docs: Optional[List[dict]] = None, mongo_db: Optional[str] = None, upsert: bool = False, collation: Optional[pymongo.collation.Collation] = None, **kwargs)[source]

Replaces many documents in a mongo collection.

Uses bulk_write with multiple ReplaceOne operations https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.bulk_write

Note

If no filter_docs``are given, it is assumed that all replacement documents contain the ``_id field which are then used as filters.

Parameters

  • mongo_collection (str) -- The name of the collection to update.

  • docs (list[dict]) -- The new documents.

  • filter_docs (list[dict]) -- A list of queries that match the documents to replace. Can be omitted; then the _id fields from docs will be used.

  • mongo_db (str) -- The name of the database to use. Can be omitted; then the database from the connection string is used.

  • upsert (bool) -- If True, perform an insert if no documents match the filters for the replace operation.

  • collation (pymongo.collation.Collation) -- An instance of Collation. This option is only supported on MongoDB 3.4 and above.

delete_one(self, mongo_collection: str, filter_doc: dict, mongo_db: Optional[str] = None, **kwargs)[source]

Deletes a single document in a mongo collection. https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.delete_one

Parameters

  • mongo_collection (str) -- The name of the collection to delete from.

  • filter_doc (dict) -- A query that matches the document to delete.

  • mongo_db (str) -- The name of the database to use. Can be omitted; then the database from the connection string is used.

delete_many(self, mongo_collection: str, filter_doc: dict, mongo_db: Optional[str] = None, **kwargs)[source]

Deletes one or more documents in a mongo collection. https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.delete_many

Parameters

  • mongo_collection (str) -- The name of the collection to delete from.

  • filter_doc (dict) -- A query that matches the documents to delete.

  • mongo_db (str) -- The name of the database to use. Can be omitted; then the database from the connection string is used.

Was this entry helpful?