Update abstra-cli

abstra_cli/apis/public.py

@@ -42,6 +42,25 @@ def usage(fname, args, kwargs, author_id, workspace_id):
         pass
 
 
+def extension(ext_session_id, api_token):
+    if os.getenv("DISABLE_USAGE_STATISTICS"):
+        return
+
+    try:
+        requests.post(
+            "https://usage-api.abstra.cloud/api/rest/cli-extension",
+            data=json.dumps(
+                {
+                    "extension_session_id": ext_session_id,
+                    "api_token": api_token,
+                }
+            ),
+            headers={"content-type": "application/json"},
+        )
+    except Exception as e:
+        pass
+
+
 def wait_for_api_token(cli_uuid):
     ws_url = f"wss://pubsub.abstra.cloud/external/sub/cli-login-{cli_uuid}"
     ws_client = None

abstra_cli/checkers.py

@@ -1,5 +1,5 @@
 import sys
-from abstra_cli.apis import get_auth_info, usage
+from abstra_cli.apis import get_auth_info, usage, extension
 
 
 def credentials_check(fname, *args, **kwargs):
@@ -17,10 +17,19 @@ def credentials_check(fname, *args, **kwargs):
 
 
 def configuration_check(fname, *args, **kwargs):
-    api_token, workspace_id, author_id = get_auth_info()
+    _, workspace_id, author_id = get_auth_info()
     usage(fname, args, kwargs, author_id, workspace_id)
 
 
 def is_logged():
-    api_token, workspace_id, author_id = get_auth_info()
+    api_token, workspace_id, _ = get_auth_info()
     return api_token and workspace_id
+
+
+def extension_check(**kwargs):
+    ext_session_id = kwargs.get("ext_session_id")
+    api_token, _, _ = get_auth_info()
+    if not ext_session_id or not api_token:
+        return
+
+    extension(ext_session_id, api_token)

abstra_cli/cli.py

@@ -37,9 +37,13 @@ def deploy(self, **kwargs):
         if not checkers.is_logged() and should_login:
             login()
 
+        checkers.extension_check(**kwargs)
         checkers.credentials_check("deploy", **kwargs)
         deploy(**kwargs)
 
+    def extension(self, **kwargs):
+        checkers.extension_check(**kwargs)
+
     def list(self, resource, **kwargs):
         checkers.credentials_check("list", resource, **kwargs)
         list_func = {