feat(ingest/aerospike): add aerospike ingestion

metadata-ingestion/src/datahub/ingestion/source/aerospike.py

@@ -1,7 +1,7 @@
 import logging
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union, ValuesView
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union, ValuesView, Set
 
 import aerospike
 import aerospike_helpers
@@ -70,7 +70,7 @@
 
 logger = logging.getLogger(__name__)
 
-DENY_NAMESPACE_LIST: set[str] = set([])
+DENY_NAMESPACE_LIST: Set[str] = set([])
 
 
 class AuthMode(Enum):
@@ -83,7 +83,7 @@ class AerospikeConfig(
     PlatformInstanceConfigMixin, EnvConfigMixin, StatefulIngestionConfigBase
 ):
     # See the Aerospike authentication docs for details and examples.
-    hosts: list[tuple] = Field(
+    hosts: List[tuple] = Field(
         default=[("localhost", 3000)], description="Aerospike hosts list."
     )
     username: Optional[str] = Field(default=None, description="Aerospike username.")
@@ -354,11 +354,10 @@ def get_field_type(
         return SchemaFieldDataType(type=TypeClass())
 
     def get_workunits_internal(self) -> Iterable[MetadataWorkUnit]:
-        sets_info: str = (
-            self.aerospike_client.info_random_node("sets")
-            .removeprefix("sets\t")
-            .removesuffix(";\n")
-        )
+        sets_info: str = self.aerospike_client.info_random_node("sets")
+        sets_info = sets_info[len("sets\t"):] if sets_info.startswith("sets\t") else sets_info
+        sets_info = sets_info[:-len(";\n")] if sets_info.endswith(";\n") else sets_info
+
         all_sets: List[AerospikeSet] = [
             AerospikeSet(item) for item in sets_info.split(";") if item
         ]