Source code for airflow.providers.amazon.aws.transfers.mongo_to_s3

#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations

import json
import warnings
from typing import TYPE_CHECKING, Any, Iterable, Sequence, cast

from bson import json_util

from airflow.models import BaseOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.mongo.hooks.mongo import MongoHook

if TYPE_CHECKING:
    from airflow.utils.context import Context


_DEPRECATION_MSG = (
    "The s3_conn_id parameter has been deprecated. You should pass instead the aws_conn_id parameter."
)


[docs]class MongoToS3Operator(BaseOperator): """Operator meant to move data from mongo via pymongo to s3 via boto. .. seealso:: For more information on how to use this operator, take a look at the guide: :ref:`howto/operator:MongoToS3Operator` :param mongo_conn_id: reference to a specific mongo connection :param aws_conn_id: reference to a specific S3 connection :param mongo_collection: reference to a specific collection in your mongo db :param mongo_query: query to execute. A list including a dict of the query :param mongo_projection: optional parameter to filter the returned fields by the query. It can be a list of fields names to include or a dictionary for excluding fields (e.g ``projection={"_id": 0}`` ) :param s3_bucket: reference to a specific S3 bucket to store the data :param s3_key: in which S3 key the file will be stored :param mongo_db: reference to a specific mongo database :param replace: whether or not to replace the file in S3 if it previously existed :param allow_disk_use: enables writing to temporary files in the case you are handling large dataset. This only takes effect when `mongo_query` is a list - running an aggregate pipeline :param compression: type of compression to use for output file in S3. Currently only gzip is supported. """
[docs] template_fields: Sequence[str] = ("s3_bucket", "s3_key", "mongo_query", "mongo_collection")
[docs] ui_color = "#589636"
[docs] template_fields_renderers = {"mongo_query": "json"}
def __init__( self, *, s3_conn_id: str | None = None, mongo_conn_id: str = "mongo_default", aws_conn_id: str = "aws_default", mongo_collection: str, mongo_query: list | dict, s3_bucket: str, s3_key: str, mongo_db: str | None = None, mongo_projection: list | dict | None = None, replace: bool = False, allow_disk_use: bool = False, compression: str | None = None, **kwargs, ) -> None: super().__init__(**kwargs) if s3_conn_id: warnings.warn(_DEPRECATION_MSG, DeprecationWarning, stacklevel=3) aws_conn_id = s3_conn_id self.mongo_conn_id = mongo_conn_id self.aws_conn_id = aws_conn_id self.mongo_db = mongo_db self.mongo_collection = mongo_collection # Grab query and determine if we need to run an aggregate pipeline self.mongo_query = mongo_query self.is_pipeline = isinstance(self.mongo_query, list) self.mongo_projection = mongo_projection self.s3_bucket = s3_bucket self.s3_key = s3_key self.replace = replace self.allow_disk_use = allow_disk_use self.compression = compression
[docs] def execute(self, context: Context): """Is written to depend on transform method""" s3_conn = S3Hook(self.aws_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=cast(list, self.mongo_query), mongo_db=self.mongo_db, allowDiskUse=self.allow_disk_use, ) else: results = MongoHook(self.mongo_conn_id).find( mongo_collection=self.mongo_collection, query=cast(dict, self.mongo_query), projection=self.mongo_projection, mongo_db=self.mongo_db, ) # Performs transform then stringifies the docs results into json format docs_str = self._stringify(self.transform(results)) s3_conn.load_string( string_data=docs_str, key=self.s3_key, bucket_name=self.s3_bucket, replace=self.replace, compression=self.compression,
) @staticmethod def _stringify(iterable: Iterable, joinable: str = "\n") -> str: """ Takes an iterable (pymongo Cursor or Array) containing dictionaries and returns a stringified version using python join """ return joinable.join([json.dumps(doc, default=json_util.default) for doc in iterable]) @staticmethod
[docs] def transform(docs: Any) -> Any: """This method is meant to be extended by child classes to perform transformations unique to those operators needs. Processes pyMongo cursor and returns an iterable with each element being a JSON serializable dictionary Base transform() assumes no processing is needed ie. docs is a pyMongo cursor of documents and cursor just needs to be passed through Override this method for custom transformations """ return docs

Was this entry helpful?