some connectors and ingestors added

querent/collectors/aws/aws_collector.py

@@ -0,0 +1,137 @@
+# import asyncio
+# from typing import AsyncGenerator
+
+# import aiofiles
+# from querent.config.collector_config import CollectorBackend, S3CollectConfig
+# from querent.collectors.collector_base import Collector
+# from querent.collectors.collector_factory import CollectorFactory
+# from querent.collectors.collector_result import CollectorResult
+# from querent.common.uri import Uri
+# import aiohttp
+# import aiobotocore
+# from aiobotocore.session import get_session
+
+
+# class AWSCollector(Collector):
+#     def __init__(self, config: S3CollectConfig):
+#         self.bucket_name = config.bucket
+#         self.region = config.region
+#         self.access_key = config.access_key
+#         self.secret_key = config.secret_key
+#         self.chunk_size = config.chunk
+
+#     # async def connect(self):
+#     #     # session = aiobotocore.get_session()
+#     #     # self.s3_client = session.create_client(
+#     #     #     's3', region_name=self.region, aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_key)
+#     #     session = aiohttp.ClientSession()
+#     #     s3_client = aiobotocore.get_session().create_client(
+#     #         's3', region_name='self.region')
+#     #     self.s3_client = s3_client
+
+#     async def connect(self):
+#         # session = aiohttp.ClientSession()
+#         session = get_session()
+#         async with session:
+#             async with session.create_client(
+#                     's3', region_name=self.region,
+#                     aws_secret_access_key=self.secret_key,
+#                     aws_access_key_id=self.access_key) as s3_client:
+#                 self.s3_client = s3_client
+
+#     async def disconnect(self):
+#         await self.session.close()
+#         await self.s3_client.close()
+
+#     async def poll(self) -> AsyncGenerator[CollectorResult, None]:
+#         async with self.s3_client.list_objects_v2(Bucket=self.bucket_name, Prefix=self.prefix) as response:
+#             for obj in response.get('Contents', []):
+#                 async with self.download_object(obj['Key']) as file:
+#                     async for chunk in self.read_chunks(file):
+#                         yield CollectorResult({"object_key": obj['Key'], "chunk": chunk})
+
+#     async def read_chunks(self, file):
+#         while True:
+#             chunk = await file.read(self.chunk_size)
+#             if not chunk:
+#                 break
+#             yield chunk
+
+#     async def download_object(self, object_key):
+#         async with aiofiles.open(object_key, 'wb') as file:
+#             await self.s3_client.download_fileobj(self.bucket_name, object_key, file)
+
+
+# class AWSCollectorFactory(CollectorFactory):
+#     def backend(self) -> CollectorBackend:
+#         return CollectorBackend.S3
+
+#     def resolve(self, uri: Uri, config: S3CollectConfig) -> Collector:
+#         config = S3CollectConfig(bucket='your_bucket_name', region='your_aws_region',
+#                                  access_key='your_access_key', secret_key='your_secret_key')
+#         return AWSCollector(config)
+
+
+import asyncio
+from typing import AsyncGenerator
+
+import aiofiles
+from querent.config.collector_config import CollectorBackend, S3CollectConfig
+from querent.collectors.collector_base import Collector
+from querent.collectors.collector_factory import CollectorFactory
+from querent.collectors.collector_result import CollectorResult
+from querent.common.uri import Uri
+import boto3
+
+
+class AWSCollector(Collector):
+    def __init__(self, config: S3CollectConfig):
+        self.bucket_name = config.bucket
+        self.region = config.region
+        self.access_key = config.access_key
+        self.secret_key = config.secret_key
+        self.chunk_size = config.chunk
+        self.s3_client = boto3.client(
+            's3',
+            aws_access_key_id=self.access_key,
+            aws_secret_access_key=self.secret_key,
+            region_name=self.region
+        )
+
+    async def connect(self):
+        pass  # No asynchronous connection needed for boto3
+
+    async def disconnect(self):
+        pass  # No asynchronous disconnect needed for boto3
+
+    async def poll(self) -> AsyncGenerator[CollectorResult, None]:
+        response = self.s3_client.list_objects_v2(
+            Bucket=self.bucket_name, Prefix=self.prefix)
+
+        for obj in response.get('Contents', []):
+            file = self.download_object(obj['Key'])
+            async for chunk in self.read_chunks(file):
+                yield CollectorResult({"object_key": obj['Key'], "chunk": chunk})
+
+    async def read_chunks(self, file):
+        while True:
+            chunk = await file.read(self.chunk_size)
+            if not chunk:
+                break
+            yield chunk
+
+    def download_object(self, object_key):
+        file_path = object_key  # Set your desired file path
+        self.s3_client.download_file(
+            self.bucket_name, object_key, file_path)
+        return open(file_path, 'rb')
+
+
+class AWSCollectorFactory(CollectorFactory):
+    def backend(self) -> CollectorBackend:
+        return CollectorBackend.S3
+
+    def resolve(self, uri: Uri, config: S3CollectConfig) -> Collector:
+        config = S3CollectConfig(bucket='pstreamsbucket1', region='ap-south-1',
+                                 access_key='AKIA5ZFZH6CA6LDWIPV5', secret_key='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')
+        return AWSCollector(config)

querent/collectors/collector_resolver.py

@@ -1,4 +1,6 @@
 from typing import Optional
+from querent.collectors.gcs.gcs_collector import GCSCollectorFactory
+from querent.collectors.aws.aws_collector import AWSCollectorFactory
 from querent.collectors.fs.fs_collector import FSCollectorFactory
 from querent.collectors.webscaper.web_scraper_collector import WebScraperFactory
 from querent.config.collector_config import CollectConfig, CollectorBackend
@@ -14,7 +16,9 @@ class CollectorResolver:
     def __init__(self):
         self.collector_factories = {
             CollectorBackend.LocalFile: FSCollectorFactory(),
+            CollectorBackend.S3: AWSCollectorFactory(),
             CollectorBackend.WebScraper: WebScraperFactory(),
+            CollectorBackend.Gcs: GCSCollectorFactory()
             # Add other collector factories as needed
         }
 
@@ -32,7 +36,13 @@ def resolve(self, uri: Uri, config: CollectConfig) -> Optional[Collector]:
     def _determine_backend(self, protocol: Protocol) -> CollectorBackend:
         if protocol.is_file_storage():
             return CollectorBackend.LocalFile
-        if protocol.is_webscrapper():
+        elif protocol.is_s3():
+            return CollectorBackend.S3
+        elif protocol.is_webscraper():
+            return CollectorBackend.WebScraper
+        elif protocol.is_gcs():
+            return CollectorBackend.Gcs
+        elif protocol.is_webscraper():
             return CollectorBackend.WebScraper
         else:
             raise CollectorResolverError(

querent/collectors/gcs/gcs_collector.py

@@ -0,0 +1,60 @@
+import asyncio
+from typing import AsyncGenerator
+
+import aiofiles
+from querent.config.collector_config import GcsCollectConfig
+from querent.config.collector_config import CollectorBackend
+from querent.collectors.collector_base import Collector
+from querent.collectors.collector_factory import CollectorFactory
+from querent.collectors.collector_result import CollectorResult
+from querent.common.uri import Uri
+import aiohttp
+from google.cloud import storage
+
+
+class GCSCollector(Collector):
+    def __init__(self, config: GcsCollectConfig):
+        self.bucket_name = config.bucket
+        self.credentials = config.credentials_path
+        self.chunk_size = config.chunk
+
+    async def connect(self):
+        self.client = storage.Client.from_service_account_json(
+            self.credentials)
+        self.bucket = self.client.get_bucket(self.bucket_name)
+
+    async def disconnect(self):
+        if self.client is not None:
+            self.client.close()
+
+    async def poll(self) -> AsyncGenerator[CollectorResult, None]:
+        blobs = self.bucket.list_blobs()
+        for blob in blobs:
+            async with self.download_blob(blob) as file:
+                async for chunk in self.read_chunks(file):
+                    yield CollectorResult({"object_key": blob.name, "chunk": chunk})
+
+    async def read_chunks(self, file):
+        while True:
+            chunk = await file.read(self.chunk_size)
+            if not chunk:
+                break
+            yield chunk
+
+    async def download_blob(self, blob):
+        file = aiofiles.open(blob.name, 'wb')
+        # Manually enter the context manager since aiofiles doesn't natively support async context management
+        await file.__aenter__()
+        with blob.open("rb") as blob_file:
+            await file.write(await blob_file.read())
+        return file
+
+
+class GCSCollectorFactory(CollectorFactory):
+    def backend(self) -> CollectorBackend:
+        return CollectorBackend.Gcs
+
+    def resolve(self, uri: Uri, config: GcsCollectConfig) -> Collector:
+        config = GcsCollectConfig(
+            bucket='your_bucket_name', credentials_path='path_to_your_credentials.json')
+        return GCSCollector(config)

querent/collectors/webscaper/web_scraper_collector.py

@@ -22,8 +22,8 @@ async def poll(self):
 
     async def scrape_website(self, website_url: str):
         content = WebpageExtractor().extract_with_bs4(website_url)
-        max_length = len(" ".join(content.split(" ")[:600]))
-        return CollectedBytes(data=content[:max_length], file=None, error=None)
+        max_length = len(' '.join(content.split(" ")[:600]))
+        return CollectedBytes(file=None, data=content[:max_length], error=None)
 
 
 class WebScraperFactory(CollectorFactory):

querent/common/uri.py

@@ -9,7 +9,7 @@
 class Protocol(enum.Enum):
     Azure = "azure"
     File = "file"
-    Grpc = "grpc"
+    Gcs = "gcs"
     PostgreSQL = "postgresql"
     Ram = "ram"
     S3 = "s3"
@@ -21,8 +21,8 @@ def is_azure(self) -> bool:
     def is_file(self) -> bool:
         return self == Protocol.File
 
-    def is_grpc(self) -> bool:
-        return self == Protocol.Grpc
+    def is_gcs(self) -> bool:
+        return self == Protocol.Gcs
 
     def is_postgresql(self) -> bool:
         return self == Protocol.PostgreSQL
@@ -42,7 +42,7 @@ def is_object_storage(self) -> bool:
     def is_database(self) -> bool:
         return self == Protocol.PostgreSQL
 
-    def is_webscrapper(self) -> bool:
+    def is_webscraper(self) -> bool:
         return self == Protocol.Webscraper
 
 
@@ -70,7 +70,7 @@ def extension(self) -> Optional[str]:
 
     @property
     def path(self) -> str:
-        return self.uri[self.protocol_idx + len(self.PROTOCOL_SEPARATOR) :]
+        return self.uri[self.protocol_idx + len(self.PROTOCOL_SEPARATOR):]
 
     def as_str(self) -> str:
         return self.uri

querent/config/collector_config.py

@@ -27,6 +27,7 @@ class S3CollectConfig(BaseModel):
     region: str
     access_key: str
     secret_key: str
+    chunk: int = 1024
 
 
 class GcsCollectConfig(BaseModel):
@@ -59,4 +60,5 @@ def from_collect_config(cls, collect_config: CollectConfig):
                 backend=CollectorBackend.WebScraper, config=WebScraperConfig()
             )
         else:
-            raise ValueError(f"Unsupported collector backend: {collect_config.backend}")
+            raise ValueError(
+                f"Unsupported collector backend: {collect_config.backend}")

querent/config/protean-tooling-368008-8b160be0bb98.json

@@ -0,0 +1,13 @@
+{
+  "type": "service_account",
+  "project_id": "protean-tooling-368008",
+  "private_key_id": "8b160be0bb98061457c793d26b4ba45572cfd35b",
+  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCqpxzT4BC+Kaiw\njoL+enKc5EPMaIU2QfxD/GvXlPHdjAN5VsxKvY3lZnmgl7ZOO1hyATbAGl9wYbNN\nZ5VkMoOMnOsN9x/SG17aywR1JxWp7X7zUCwq4sHa8IIscSmJKkY0nqdcC+3w9H3I\nCW3x7Zmx0Rdpq5fhP86h1/7A/xcURghRu9xoTMy72wco1JLPS/vDTrNV6WJujfP/\n+u9uyF6zm7UzPng3kNRK0eM6ePAHcV8+v0WzhsGhF2v9qNzr17J4D5xFO2EX0V/S\na2PVIJj4OhECCK9zOQfFamyb5JlPd+aq0VWxSeyR/tQUHGBtwPZZj9XVWGjU1Txe\nbeZJY8XhAgMBAAECggEASYKD5fa6b5fVXPx+i+0mCtQ/KkGJxHYHGygUMkCsMSF7\na5zKkpV1lrAeS2Geu7NPpYhFU9noa6oCSSEKFqe1x+/eRt1BXUnCAeB68pYb/F9c\nFKZf6DdWhx+cC1XRrQ7DQC5pbDqNTAKdXfpF3HVbg9/MRI4Qdgx8z3WR/QPFJstR\n5KbF7WhXHIL2sA5AAHYamQwcUqG2/1ci67uxjJb08dvPkkIrI3A/0ZDWI8p9MC/h\naptl2MEwIwZDxm03iqwqzhqhDOuIbMGPiG6qZHRkzbjAij3FcX4vYLVXV3KeOV7u\n4vuTZTWRqG1TefVNstexGMpgxKSTfMvC0HJTf0QQmwKBgQDYKuhBmdnMrxm7E5pk\nrnbTMez0HPoowX5W9T2T1NY3luTFDSWejwucyAmbLEKZEa0hv7SMVxuWUCVZeGKb\nTS7tuNEu5Bzd3vZg2m2wMSdLTyO6NcozpXwE8+t9/BCdGh2EXojwOq/ntx/ADuJ2\nzninIlM2zkmX8bXXRmiw8hHaZwKBgQDKGSxCYqA1phfdcjMaGDwrEJfgUt1a8A0Y\nOu1Vy42wunjMgEYPimRsgv1QpFb8RG2XsqREcXhcIgliTQrlaSQuuF3AgdYmUyPI\nqfBtbci/9awbb/oOytgqu6cveRWjL5kNUOhtufOXZIbkCbeL7+55zgTFA5vclhqm\ntsot7CbAdwKBgB50RQ5g73dPv+9EPr8qeSEd6oUlw/ST9gmoOUmndZqVTpuYU5F4\nE6m94OSUorlP7TVQ/c27w+40Ul69gB7jC9Z/8DCMAKkdbVaNz2BAPIn2l1D3nQoT\nqVpFRwsTxP9tG3V3oBFAI2IgwX9cWXllc0gQpOqWu3WFnxm6GKu9HtjHAoGAMPdu\ncNtCpB0NNTuBRDZRBcIsek65kVoOYiuM67e5KfPUZwBBeWx3ro8CX3KiHNdbkmDX\nSjoQBoq8qVOvayxRdvx2wrjysTB9wBXkyi7KcDBPNgqolMLzI096T5dP9c0HkWFk\nH4npimnyAPiVv1Mv/dU79JxOPGEmLfSmAgBw1zsCgYBmGUh2wnFBNXltJfDKA/fL\n+Q+yXKIiX7UO7XYRenxz/XYH3qEkDAVJsWdFEt3vs8YBJKg3gewzGbhnyWv3udSi\n8oiE0Om/tMyHmMTAlmRM79RFsmdyFRZ20ckb/EwW35L74RFTM2HybCl0ax4kuDQH\nWcy7orKon24IyeIeKpcF3A==\n-----END PRIVATE KEY-----\n",
+  "client_email": "pstreams1@protean-tooling-368008.iam.gserviceaccount.com",
+  "client_id": "108934245481292748227",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://oauth2.googleapis.com/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/pstreams1%40protean-tooling-368008.iam.gserviceaccount.com",
+  "universe_domain": "googleapis.com"
+}

querent/connectors/aws_connector.py

@@ -0,0 +1,70 @@
+import boto3
+from botocore.exceptions import NoCredentialsError, PartialCredentialsError, BotoCoreError
+
+def initialize_s3_resource():
+    try:
+        s3_resource = boto3.resource('s3',
+                                     aws_access_key_id='YOUR_ACCESS_KEY',
+                                     aws_secret_access_key='YOUR_SECRET_KEY',
+                                     region_name='YOUR_REGION')
+        return s3_resource
+    except (PartialCredentialsError, BotoCoreError) as e:
+        print(f"Error initializing S3 resource: {e}")
+        return None
+
+def list_objects(bucket_name):
+    try:
+        s3_resource = initialize_s3_resource()
+        if s3_resource is None:
+            return []
+
+        bucket = s3_resource.Bucket(bucket_name)
+        objects = list(bucket.objects.all())
+        return objects
+    except BotoCoreError as e:
+        print(f"Error listing objects: {e}")
+        return []
+
+def download_file(bucket_name, s3_key, local_path):
+    try:
+        s3_resource = initialize_s3_resource()
+        if s3_resource is None:
+            return False
+
+        s3_resource.Bucket(bucket_name).download_file(s3_key, local_path)
+        return True
+    except (BotoCoreError, NoCredentialsError) as e:
+        print(f"Error downloading file: {e}")
+        return False
+
+def upload_file(bucket_name, s3_key, local_path):
+    try:
+        s3_resource = initialize_s3_resource()
+        if s3_resource is None:
+            return False
+
+        s3_resource.Bucket(bucket_name).upload_file(local_path, s3_key)
+        return True
+    except (BotoCoreError, NoCredentialsError) as e:
+        print(f"Error uploading file: {e}")
+        return False
+
+def delete_object(bucket_name, s3_key):
+    try:
+        s3_resource = initialize_s3_resource()
+        if s3_resource is None:
+            return False
+
+        s3_resource.Object(bucket_name, s3_key).delete()
+        return True
+    except (BotoCoreError, NoCredentialsError) as e:
+        print(f"Error deleting object: {e}")
+        return False
+
+# Usage
+bucket_name = 'your-s3-bucket-name'
+object_list = list_objects(bucket_name)
+for obj in object_list:
+    print(obj.key)
+
+# Remember to replace 'YOUR_ACCESS_KEY', 'YOUR_SECRET_KEY', and 'YOUR_REGION' with your actual credentials.