airflow/contrib/operators/mongo_to_s3.py
def execute(self, context):
        """
        Executed by task_instance at runtime
        """
        s3_conn = S3Hook(self.s3_conn_id)

        # Grab collection and execute query according to whether or not it is a pipeline
        if self.is_pipeline:
            results = MongoHook(self.mongo_conn_id).aggregate(
                mongo_collection=self.mongo_collection,
                aggregate_query=self.mongo_query,
                mongo_db=self.mongo_db
            )

        else:
            results = MongoHook(self.mongo_conn_id).find(
                mongo_collection=self.mongo_collection,
                query=self.mongo_query,
                mongo_db=self.mongo_db
            )

        # Performs transform then stringifies the docs results into json format
        docs_str = self._stringify(self.transform(results))

        # Load Into S3
        s3_conn.load_string(
            string_data=docs_str,
            key=self.s3_key,
            bucket_name=self.s3_bucket,
            replace=self.replace
        )

        return True
def execute(self, context):
        s3_conn = S3Hook(self.s3_conn_id)

        # Grab collection and execute query according to whether or not it is a pipeline
        if self.is_pipeline:
            results = MongoHook(self.mongo_conn_id).aggregate(
                mongo_collection=self.mongo_collection,
                aggregate_query=self.mongo_query,
                mongo_db=self.mongo_db
            )

        else:
            results = MongoHook(self.mongo_conn_id).find(
                mongo_collection=self.mongo_collection,
                query=self.mongo_query,
                mongo_db=self.mongo_db
            )

        # Performs transform then stringifies the docs results into json format
        docs_str = self._stringify(self.transform(results))

        # Load Into S3
        s3_conn.load_string(
            string_data=docs_str,
            key=self.s3_key,
            bucket_name=self.s3_bucket,
            replace=self.replace
        )

        return True
def execute ( self , context ) : s3_conn = S3Hook ( self . s3_conn_id ) # Grab collection and execute query according to whether or not it is a pipeline if self . is_pipeline : results = MongoHook ( self . mongo_conn_id ) . aggregate ( mongo_collection = self . mongo_collection , aggregate_query = self . mongo_query , mongo_db = self . mongo_db ) else : results = MongoHook ( self . mongo_conn_id ) . find ( mongo_collection = self . mongo_collection , query = self . mongo_query , mongo_db = self . mongo_db ) # Performs transform then stringifies the docs results into json format docs_str = self . _stringify ( self . transform ( results ) ) # Load Into S3 s3_conn . load_string ( string_data = docs_str , key = self . s3_key , bucket_name = self . s3_bucket , replace = self . replace ) return True
Executed by task_instance at runtime
Executed by task_instance at runtime