Add cli:

Get list of uniue addresses
Add subscription to moonworm tasks journal

backend/moonstreamapi/actions.py

@@ -8,7 +8,7 @@ from enum import Enum
 import uuid
 
 import boto3  # type: ignore
-from bugout.data import BugoutSearchResults
+from bugout.data import BugoutSearchResults, BugoutSearchResult
 from bugout.journal import SearchOrder
 from ens.utils import is_valid_ens_name  # type: ignore
 from eth_utils.address import is_address  # type: ignore
@@ -438,13 +438,13 @@ def upload_abi_to_s3(
 
 def get_all_entries_from_search(
     journal_id: str, search_query: str, limit: int, token: str
-) -> List[Any]:
+) -> List[BugoutSearchResult]:
     """
     Get all required entries from journal using search interface
     """
     offset = 0
 
-    results: List[Any] = []
+    results: List[BugoutSearchResult] = []
 
     try:
         existing_metods = bc.search(

backend/moonstreamapi/admin/cli.py

@@ -118,7 +118,7 @@ def migrations_run(args: argparse.Namespace) -> None:
 
 def moonworm_tasks_list_handler(args: argparse.Namespace) -> None:
 
-    moonworm_tasks.get_list_of_tags(args.query, args.tag)
+    moonworm_tasks.get_list_of_address()
 
 
 def moonworm_tasks_add_subscription_handler(args: argparse.Namespace) -> None:
@@ -358,23 +358,7 @@ This CLI is configured to work with the following API URLs:
         "list", description="Return list of addresses in moonworm journal."
     )
 
-    parser_moonworm_tasks_list.add_argument(
-        "-q",
-        "--query",
-        type=str,
-        help="query filter.",
-    )
-
-    parser_moonworm_tasks_list.add_argument(
-        "-t",
-        "--tag",
-        default="address",
-        choices=["address"],
-        type=str,
-        help="Tag for wich we fetch and return values.",
-    )
-
-    parser_moonworm_tasks.set_defaults(func=moonworm_list_handler)
+    parser_moonworm_tasks_list.set_defaults(func=moonworm_tasks_list_handler)
 
     parser_moonworm_tasks_add = subcommands_moonworm_tasks.add_parser(
         "add_subscription", description="Manage tasks for moonworm journal."
@@ -387,7 +371,7 @@ This CLI is configured to work with the following API URLs:
         help="Id of subscription for add to moonworm tasks.",
     )
 
-    parser_moonworm_tasks.set_defaults(func=moonworm_task)
+    parser_moonworm_tasks_add.set_defaults(func=moonworm_tasks_add_subscription_handler)
 
     args = parser.parse_args()
     args.func(args)

backend/moonstreamapi/admin/moonworm_tasks.py

@@ -1,23 +1,86 @@
 import logging
+import json
 
-from ..settings import BUGOUT_REQUEST_TIMEOUT_SECONDS, MOONSTREAM_ADMIN_ACCESS_TOKEN, 
+import boto3
+from bugout.data import BugoutResource, BugoutResources
+from bugout.exceptions import BugoutResponseException
+
+
+from ..actions import get_all_entries_from_search, apply_moonworm_tasks
+from ..settings import MOONSTREAM_ADMIN_ACCESS_TOKEN, MOONSTREAM_MOONWORM_TASKS_JOURNAL
 from ..settings import bugout_client as bc
 
 logger = logging.getLogger(__name__)
 
 
+def get_list_of_addresses():
+    """
+    Return list of addresses of tasks
+    """
 
-def get_list_of_tags(query: str, tag: str):
+    entries = get_all_entries_from_search(
+        journal_id=MOONSTREAM_MOONWORM_TASKS_JOURNAL,
+        search_query=f"",
+        limit=100,
+        token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+    )
+
+    addresses = set()
+
+    for entry in entries:
+
+        addresses.add(entry.title)
+
+    print(addresses)
+
+
+def add_subscription(id: str):
     """
     Return list of tags depends on query and tag
     """
 
-    existing_metods = bc.search(
-        token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
-        journal_id=journal_id,
-        query=search_query,
-        content=False,
-        timeout=10.0,
-        limit=limit,
-        offset=offset,
-    )
+    try:
+        subscription_resource: BugoutResource = bc.get_resource(
+            token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
+            resource_id=id,
+        )
+
+    except BugoutResponseException as e:
+        logging.error(f"Bugout error: {str(e)}")
+    except Exception as e:
+        logger.error(f"Error get resource: {str(e)}")
+
+    s3_client = boto3.client("s3")
+
+    if subscription_resource.resource_data["abi"] is not None:
+        logger.error(f"Resource don't have abi.")
+
+        bucket = subscription_resource.resource_data["bucket"]
+        key = subscription_resource.resource_data["s3_path"]
+
+        if bucket is None or key is None:
+            logger.error(f"Error subscription not have s3 path to abi")
+
+        s3_path = f"s3://{bucket}/{key}"
+
+        try:
+
+            response = s3_client.get_object(
+                Bucket=bucket,
+                Key=key,
+            )
+
+        except s3_client.exceptions.NoSuchKey as e:
+            logger.error(
+                f"Error getting Abi for subscription {str(subscription_resource.subscription_id)} S3 {s3_path} does not exist : {str(e)}"
+            )
+
+        abi = json.loads(response["Body"].read())
+
+        apply_moonworm_tasks(
+            subscription_type=subscription_resource.resource_data["type"],
+            abi=abi,
+            address=subscription_resource.resource_data["address"],
+        )
+    else:
+        logging.info("For apply to moonworm tasks subscriptions must have an abi.")