airflow_dag_mongodb_document_insert_with_mongo_operator.py

1from datetime import datetime
2from airflow import DAG
3from airflow.providers.mongo.operators.mongo import MongoOperator
4
5with DAG(
6    "example_mongo_operator",
7    start_date=datetime(2022, 1, 1),
8    schedule=None,
9    catchup=False,
10) as dag:
11
12    # Example task to insert a document into a collection
13    # Note: 'mongo_default' is the default connection ID
14    insert_document = MongoOperator(
15        task_id="insert_document",
16        mongo_conn_id="mongo_default",
17        collection="test_collection",
18        mongo_db="test_db",
19        command={
20            "insert": "test_collection",
21            "documents": [{"name": "John Doe", "email": "john.doe@example.com"}]
22        },
23    )
24
25    insert_document