diff --git a/plugins/loaders/target-jsonl--andyh1203.lock b/plugins/loaders/target-jsonl--andyh1203.lock new file mode 100644 index 0000000..11fa0ba --- /dev/null +++ b/plugins/loaders/target-jsonl--andyh1203.lock @@ -0,0 +1,34 @@ +{ + "plugin_type": "loaders", + "name": "target-jsonl", + "namespace": "target_jsonl", + "variant": "andyh1203", + "label": "JSON Lines (JSONL)", + "docs": "https://hub.meltano.com/loaders/target-jsonl--andyh1203", + "repo": "https://github.com/andyh1203/target-jsonl", + "pip_url": "target-jsonl", + "description": "JSONL loader", + "logo_url": "https://hub.meltano.com/assets/logos/loaders/jsonl.png", + "settings": [ + { + "name": "destination_path", + "kind": "string", + "value": "output", + "label": "Destination Path", + "description": "Sets the destination path the JSONL files are written to, relative\nto the project root.\n\nThe directory needs to exist already, it will not be created\nautomatically.\n\nTo write JSONL files to the project root, set an empty string (`\"\"`).\n" + }, + { + "name": "do_timestamp_file", + "kind": "boolean", + "value": false, + "label": "Include Timestamp in File Names", + "description": "Specifies if the files should get timestamped.\n\nBy default, the resulting file will not have a timestamp in the file name (i.e. `exchange_rate.jsonl`).\n\nIf this option gets set to `true`, the resulting file will have a timestamp associated with it (i.e. `exchange_rate-{timestamp}.jsonl`).\n" + }, + { + "name": "custom_name", + "kind": "string", + "label": "Custom File Name Override", + "description": "Specifies a custom name for the filename, instead of the stream name.\n\nThe file name will be `{custom_name}-{timestamp}.jsonl`, if `do_timestamp_file` is `true`.\nOtherwise the file name will be `{custom_name}.jsonl`.\n\nIf custom name is not provided, the stream name will be used.\n" + } + ] +} diff --git a/pyproject.toml b/pyproject.toml index 6395c88..30ab036 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "tap-mongodb" -version = "2.3.3" +version = "2.4.0" description = "`tap-mongodb` is a Singer tap for MongoDB and AWS DocumentDB, built with the Meltano Singer SDK." readme = "README.md" authors = ["Matt Menzenski"] diff --git a/tap_mongodb/connector.py b/tap_mongodb/connector.py index 366192a..ca6f35b 100644 --- a/tap_mongodb/connector.py +++ b/tap_mongodb/connector.py @@ -2,7 +2,7 @@ import sys from logging import Logger, getLogger -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from pymongo import MongoClient from pymongo.database import Database @@ -17,6 +17,15 @@ from functools import cached_property +try: + from typing import TypeAlias # pylint: disable=ungrouped-imports + + MongoVersion: TypeAlias = Tuple[int, int] +except ImportError: + TypeAlias = None + MongoVersion = Tuple[int, int] + + class MongoDBConnector: """MongoDB/DocumentDB connector class""" @@ -34,6 +43,7 @@ def __init__( # pylint: disable=too-many-arguments self._datetime_conversion: str = datetime_conversion.upper() self._prefix: Optional[str] = prefix self._logger: Logger = getLogger(__name__) + self._version: Optional[MongoVersion] = None @cached_property def mongo_client(self) -> MongoClient: @@ -42,7 +52,9 @@ def mongo_client(self) -> MongoClient: self._connection_string, datetime_conversion=self._datetime_conversion, **self._options ) try: - client.server_info() + server_info: Dict[str, Any] = client.server_info() + version_array: List[int] = server_info["versionArray"] + self._version = (version_array[0], version_array[1]) except Exception as exception: self._logger.exception("Could not connect to MongoDB") raise RuntimeError("Could not connect to MongoDB") from exception @@ -53,6 +65,11 @@ def database(self) -> Database: """Provide a Database instance.""" return self.mongo_client[self._db_name] + @property + def version(self) -> Optional[MongoVersion]: + """Returns the MongoVersion that is being used.""" + return self._version + def get_fully_qualified_name( self, collection_name: str, diff --git a/tap_mongodb/streams.py b/tap_mongodb/streams.py index 2232920..21d0bf0 100644 --- a/tap_mongodb/streams.py +++ b/tap_mongodb/streams.py @@ -218,7 +218,13 @@ def get_records(self, context: dict | None) -> Iterable[dict]: change_stream_options = {"full_document": "updateLookup"} if bookmark is not None and bookmark != DEFAULT_START_DATE: self.logger.debug(f"using bookmark: {bookmark}") - change_stream_options["resume_after"] = {"_data": bookmark} + # if on mongo version 4.2 or above, use start_after instead of resume_after, as the former will + # gracefully open a new change stream if the resume token's event is not present in the oplog, while + # the latter will error in that scenario. + if self._connector.version >= (4, 2): + change_stream_options["start_after"] = {"_data": bookmark} + else: + change_stream_options["resume_after"] = {"_data": bookmark} operation_types_allowlist: set = set(self.config.get("operation_types")) has_seen_a_record: bool = False keep_open: bool = True @@ -245,7 +251,8 @@ def get_records(self, context: dict | None) -> Iterable[dict]: f"Unable to enable change streams on collection {collection.name}" ) from operation_failure elif ( - operation_failure.code == 286 + self._connector.version < (4, 2) + and operation_failure.code == 286 and "as the resume point may no longer be in the oplog." in operation_failure.details["errmsg"] ): self.logger.warning("Unable to resume change stream from resume token. Resetting resume token.") @@ -265,7 +272,8 @@ def get_records(self, context: dict | None) -> Iterable[dict]: record = change_stream.try_next() except OperationFailure as operation_failure: if ( - operation_failure.code == 286 + self._connector.version < (4, 2) + and operation_failure.code == 286 and "as the resume point may no longer be in the oplog." in operation_failure.details["errmsg"] ):