Using MongoDB with Apache Airflow

1
import os
2
import json
3
from airflow import DAG
4
from airflow.operators.python import PythonOperator
5
from airflow.operators.bash import BashOperator
6
from airflow.providers.http.operators.http import SimpleHttpOperator
7
from airflow.providers.mongo.hooks.mongo import MongoHook
8
from datetime import datetime,timedelta
9
10
def on_failure_callback(**context):
11
    print(f"Task {context['task_instance_key_str']} failed.")
12
13
def uploadtomongo(ti, **context):
14
    try:
15
        hook = MongoHook(mongo_conn_id='mongoid')
16
        client = hook.get_conn()
17
        db = client.MyDB
18
        currency_collection=db.currency_collection
19
        print(f"Connected to MongoDB - {client.server_info()}")
20
        d=json.loads(context["result"])
21
        currency_collection.insert_one(d)
22
    except Exception as e:
23
        printf("Error connecting to MongoDB -- {e}")
24
25
with DAG(
26
    dag_id="load_currency_data",
27
    schedule_interval=None,
28
    start_date=datetime(2022,10,28),
29
    catchup=False,
30
    tags= ["currency"],
31
    default_args={
32
        "owner": "Rob",
33
        "retries": 2,
34
        "retry_delay": timedelta(minutes=5),
35
        'on_failure_callback': on_failure_callback
36
    }
37
) as dag:
38
39
    t1 = SimpleHttpOperator(
40
        task_id='get_currency',
41
        method='GET',
42
        endpoint='2022-01-01..2022-06-30',
43
        headers={"Content-Type": "application/json"},
44
        do_xcom_push=True,
45
        dag=dag)
46
47
    t2 = PythonOperator(
48
        task_id='upload-mongodb',
49
        python_callable=uploadtomongo,
50
        op_kwargs={"result": t1.output},
51
        dag=dag
52
        )
53
54
    t1 >> t2

1
t1 = SimpleHttpOperator(
2
        task_id='get_currency',
3
        method='GET',
4
        endpoint='2022-01-01..2022-06-30',
5
        headers={"Content-Type": "application/json"},
6
        do_xcom_push=True,
7
        dag=dag)

1
t2 = PythonOperator(
2
        task_id='upload-mongodb',
3
        python_callable=uploadtomongo,
4
        op_kwargs={"result": t1.output},
5
        dag=dag
6
        )

1
def uploadtomongo(ti, **context):
2
    try:
3
        hook = MongoHook(mongo_conn_id='mongoid')
4
        client = hook.get_conn()
5
        db = client.MyDB
6
        currency_collection=db.currency_collection
7
        print(f"Connected to MongoDB - {client.server_info()}")
8
        d=json.loads(context["result"])
9
        currency_collection.insert_one(d)
10
    except Exception as e:
11
        printf("Error connecting to MongoDB -- {e}")

1
t1 >> t2

1
with DAG(
2
    dag_id="load_currency_data",
3
    schedule=None,
4
    start_date=datetime(2022,10,28),
5
    catchup=False,
6
    tags= ["currency"],
7
    default_args={
8
        "owner": "Rob",
9
        "retries": 2,
10
        "retry_delay": timedelta(minutes=5),
11
        'on_failure_callback': on_failure_callback
12
    }
13
) as dag:

1
{
2
    _id: ObjectId("635b25bdcef2d967af053e2c"),
3
    amount: 1,
4
    base: 'EUR',
5
    start_date: '2022-01-03',
6
    end_date: '2022-06-30',
7
    rates: {
8
      '2022-01-03': {
9
        AUD: 1.5691,
10
        BGN: 1.9558,
11
        BRL: 6.3539,
12
… },
13
},
14
      '2022-01-04': {
15
        AUD: 1.5682,
16
        BGN: 1.9558,
17
        BRL: 6.4174,
18
… }

1
[{$project:{
2
	rates:{
3
		$objectToArray:"$rates"}}},{
4
			$unwind:"$rates"
5
		}
6
		,{
7
			$project:{
8
				_id:0,"date":"$rates.k","Value":"$rates.v"}}]

1
{
2
    date: '2022-01-03',
3
    Value: {
4
      AUD: 1.5691,
5
      BGN: 1.9558,
6
      BRL: 6.3539,
7
…  },
8
  {
9
    date: '2022-01-04',
10
    Value: {
11
      AUD: 1.5682,
12
      BGN: 1.9558,
13
      BRL: 6.4174,
14
..}
15
}