Delete More Stuff (#52)

* delete non-generic stuffs

* add todo with link to flashbot upload (#1)

* add todo with link to flashbot upload

* whirlwind of refactor

* add files

* delete more

.env.sample

@@ -1,10 +1,3 @@
-VOLUME_PATH=data
-APP_DATA_MAX_RETRIES=3
-APP_DATA_GIVE_UP_THRESHOLD=100
-
-# Dune credentials
-DUNE_API_KEY=
-
 # AWS Credentials
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
@@ -13,10 +6,3 @@ AWS_INTERNAL_ROLE=
 AWS_EXTERNAL_ROLE=
 AWS_EXTERNAL_ID=
 AWS_BUCKET=
-
-#Orderbook DB Credentials
-BARN_DB_URL={user}:{password}@{host}:{port}/{database}
-PROD_DB_URL={user}:{password}@{host}:{port}/{database}
-
-# IPFS Gateway
-IPFS_ACCESS_TOKEN=

.github/workflows/pull-request.yaml

@@ -24,29 +24,3 @@ jobs:
         run: black --check ./
       - name: Type Check (mypy)
         run: mypy src --strict
-
-  tests:
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        python-version: ["3.10", "3.11"]
-
-    steps:
-      - uses: actions/checkout@v2
-      - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v2
-        with:
-          python-version: ${{ matrix.python-version }}
-
-      - name: Install Requirements
-        run: pip install -r requirements/dev.txt
-      - name: Unit Tests
-        run: python -m pytest tests/unit
-        env:
-          IPFS_ACCESS_KEY: ${{ secrets.IPFS_ACCESS_KEY }}
-      - name: Integration Tests
-        run: python -m pytest tests/integration
-        env:
-          PROD_DB_URL: ${{ secrets.PROD_DB_URL }}
-          BARN_DB_URL: ${{ secrets.BARN_DB_URL }}
-          WAREHOUSE_URL: ${{ secrets.WAREHOUSE_URL }}

requirements/dev.txt

@@ -5,5 +5,4 @@ black>=22.6.0
 mypy==1.3.0
 mypy-extensions==1.0.0
 pylint>=2.14.4
-pytest>=7.1.2
-sqlalchemy-stubs>=0.4
+pytest>=7.1.2

requirements/prod.txt

@@ -1,9 +1,5 @@
 dune-client>=0.3.0
-psycopg2-binary>=2.9.3
 python-dotenv>=0.20.0
 requests>=2.28.1
 pandas>=1.5.0
-ndjson>=0.3.1
-py-multiformats-cid>=0.4.4
-boto3>=1.26.12
-SQLAlchemy<2.0
+boto3>=1.26.12

src/post/aws.py → src/aws.py

@@ -1,6 +1,7 @@
 """Aws S3 Bucket functionality (namely upload_file)"""
 from __future__ import annotations
 
+import json
 import os
 from collections import defaultdict
 from dataclasses import dataclass
@@ -13,7 +14,7 @@
 from dotenv import load_dotenv
 
 from src.logger import set_log
-from src.models.tables import SyncTable
+from src.text_io import BytesIteratorIO
 
 log = set_log(__name__)
 
@@ -78,21 +79,17 @@ def from_bucket_collection(cls, bucket_objects: Any) -> BucketStructure:
             object_key = bucket_obj.key
             path, _ = object_key.split("/")
             grouped_files[path].append(BucketFileObject.from_key(object_key))
-            if path not in SyncTable.supported_tables():
-                # Catches any unrecognized filepath.
-                log.warning(f"Found unexpected file {object_key}")
 
         log.debug(f"loaded bucket filesystem: {grouped_files.keys()}")
 
         return cls(files=grouped_files)
 
-    def get(self, table: SyncTable | str) -> list[BucketFileObject]:
+    def get(self, table: str) -> list[BucketFileObject]:
         """
         Returns the list of files under `table`
             - returns empty list if none available.
         """
-        table_str = str(table) if isinstance(table, SyncTable) else table
-        return self.files.get(table_str, [])
+        return self.files.get(table, [])
 
 
 class AWSClient:
@@ -131,16 +128,22 @@ def _assume_role(self) -> ServiceResource:
         """
         sts_client = boto3.client("sts")
 
+        # TODO - assume that the internal role is already assumed. and use get session_token
+        # sts_client.get_session_token()
         internal_assumed_role_object = sts_client.assume_role(
             RoleArn=self.internal_role,
             RoleSessionName="InternalSession",
         )
         credentials = internal_assumed_role_object["Credentials"]
+        # sts_client.get_session_token()
+
         sts_client = boto3.client(
             "sts",
-            aws_access_key_id=credentials["AccessKeyId"],
-            aws_secret_access_key=credentials["SecretAccessKey"],
-            aws_session_token=credentials["SessionToken"],
+            aws_access_key_id=credentials["AccessKeyId"],  # AWS_ACCESS_KEY_ID
+            aws_secret_access_key=credentials[
+                "SecretAccessKey"
+            ],  # AWS_SECRET_ACCESS_KEY
+            aws_session_token=credentials["SessionToken"],  # AWS_SESSION_TOKEN
         )
 
         external_assumed_role_object = sts_client.assume_role(
@@ -176,14 +179,37 @@ def upload_file(self, filename: str, object_key: str) -> bool:
         log.debug(f"uploaded {filename} to {self.bucket}")
         return True
 
+    def put_object(self, data_set: list[dict[str, Any]], object_key: str) -> bool:
+        """Upload a file to an S3 bucket
+
+        :param data_list: Data to upload. Should be a full path to file.
+        :param object_key: S3 object key. For our purposes, this would
+                           be f"{table_name}/cow_{latest_block_number}.json"
+        :return: True if file was uploaded, else raises
+        """
+
+        file_object = BytesIteratorIO(
+            f"{json.dumps(row)}\n".encode("utf-8") for row in data_set
+        )
+
+        s3_client = self._get_s3_client(self._assume_role())
+
+        s3_client.upload_fileobj(
+            file_object,
+            bucket=self.bucket,
+            key=object_key,
+            extra_args={"ACL": "bucket-owner-full-control"},
+        )
+        log.debug(f"uploaded {object_key} to {self.bucket}")
+        return True
+
     def delete_file(self, object_key: str) -> bool:
         """Delete a file from an S3 bucket
 
         :param object_key: S3 object key. For our purposes, this would
                            be f"{table_name}/cow_{latest_block_number}.json"
         :return: True if file was deleted, else raises
         """
-        # TODO - types! error: "BaseClient" has no attribute "delete_object"
         s3_client = self._get_s3_client(self._assume_role())
         s3_client.delete_object(  # type: ignore
             Bucket=self.bucket,
@@ -220,22 +246,21 @@ def existing_files(self) -> BucketStructure:
         bucket_objects = bucket.objects.all()
         return BucketStructure.from_bucket_collection(bucket_objects)
 
-    def last_sync_block(self, table: SyncTable | str) -> int:
+    def last_sync_block(self, table: str) -> int:
         """
         Based on the existing bucket files,
         the last sync block is uniquely determined from the file names.
         """
-        table_str = str(table) if isinstance(table, SyncTable) else table
         try:
-            table_files = self.existing_files().get(table_str)
+            table_files = self.existing_files().get(table)
             return max(file_obj.block for file_obj in table_files if file_obj.block)
         except ValueError as err:
             # Raised when table_files = []
             raise FileNotFoundError(
                 f"Could not determine last sync block for {table} files. No files."
             ) from err
 
-    def delete_all(self, table: SyncTable | str) -> None:
+    def delete_all(self, table: str) -> None:
         """Deletes all files within the supported tables directory"""
         log.info(f"Emptying Bucket {table}")
         try:
@@ -245,6 +270,4 @@ def delete_all(self, table: SyncTable | str) -> None:
                 log.info(f"Deleting file {file_data.object_key}")
                 self.delete_file(file_data.object_key)
         except KeyError as err:
-            raise ValueError(
-                f"Invalid table_name {table}, please chose from {SyncTable.supported_tables()}"
-            ) from err
+            raise ValueError(f"invalid table name {table}") from err

src/environment.py

@@ -2,5 +2,4 @@
 from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).parent.parent
-QUERY_PATH = PROJECT_ROOT / Path("src/sql")
 LOG_CONFIG_FILE = PROJECT_ROOT / Path("logging.conf")