rippling cli : Flux app command

rippling_cli/cli/commands/flux/app.py

@@ -0,0 +1,67 @@
+import os
+
+import click
+
+from rippling_cli.config.config import get_app_config, save_app_config
+from rippling_cli.constants import RIPPLING_API
+from rippling_cli.core.api_client import APIClient
+from rippling_cli.utils.login_utils import ensure_logged_in
+
+
+@click.group()
+@click.pass_context
+def app(ctx: click.Context) -> None:
+    """Manage flux apps"""
+    ensure_logged_in(ctx)
+
+
+@app.command()
+def list() -> None:
+    """This command displays a list of all apps owned by the developer."""
+    ctx: click.Context = click.get_current_context()
+    api_client = APIClient(base_url=RIPPLING_API, headers={"Authorization": f"Bearer {ctx.obj.oauth_token}"})
+    endpoint = "/apps/api/integrations"
+
+    for page in api_client.find_paginated(endpoint):
+        click.echo(f"Page: {len(page)} apps")
+
+        for app in page:
+            click.echo(f"- {app.get('displayName')} ({app.get('id')})")
+
+        if not click.confirm("Continue"):
+            break
+
+    click.echo("End of apps list.")
+
+
+@app.command()
+@click.option("--app_id", required=True, type=str, help="The app id to set for the current directory.")
+def set(app_id: str) -> None:
+    """This command sets the current app within the app_config.json file located in the .rippling directory."""
+    ctx: click.Context = click.get_current_context()
+    api_client = APIClient(base_url=RIPPLING_API, headers={"Authorization": f"Bearer {ctx.obj.oauth_token}"})
+
+    endpoint = "/apps/api/apps/?large_get_query=true"
+    response = api_client.post(endpoint, data={"query": f"id={app_id}&limit=1"})
+    app_list = response.json() if response.status_code == 200 else []
+
+    if response.status_code != 200 or len(app_list) == 0:
+        click.echo(f"Invalid app id: {app_id}")
+        return
+
+    app_name = app_list[0].get("displayName")
+
+    save_app_config(app_id, app_name)
+    click.echo(f"Current app set to {app_name} ({app_id})")
+
+
+@app.command()
+def current() -> None:
+    """This command indicates the current app selected by the developer within the directory."""
+    app_config_json = get_app_config()
+    app_config = app_config_json.get(os.getcwd())
+    if not app_config:
+        click.echo("No app selected.")
+        return
+
+    click.echo(f"{app_config.get('displayName')} ({app_config.get('id')})")

rippling_cli/cli/commands/flux/flux.py

@@ -0,0 +1,14 @@
+import click
+
+from rippling_cli.cli.commands.flux.app import app
+from rippling_cli.utils.login_utils import ensure_logged_in
+
+
+@click.group()
+@click.pass_context
+def flux(ctx: click.Context) -> None:
+    """Manage flux apps"""
+    ensure_logged_in(ctx)
+
+
+flux.add_command(app)  # type: ignore

rippling_cli/cli/main.py

@@ -3,6 +3,7 @@
 
 import click
 
+from rippling_cli.cli.commands.flux.flux import flux
 from rippling_cli.cli.commands.login import login
 from rippling_cli.config.config import get_client_id, get_oauth_token_data
 from rippling_cli.constants import EXIT_UNKNOWN_EXCEPTION
@@ -41,7 +42,8 @@ def cli(ctx):
 
 
 COMMANDS_LIST: list[Union[click.Command, click.Group]] = [
-    login
+    login,
+    flux
 ]
 
 

rippling_cli/config/config.py

@@ -5,7 +5,7 @@
 from datetime import datetime, timedelta
 from pathlib import Path
 
-from rippling_cli.constants import OAUTH_TOKEN_FILE_NAME, RIPPLING_DIRECTORY_NAME
+from rippling_cli.constants import APP_CONFIG_FILE, OAUTH_TOKEN_FILE_NAME, RIPPLING_DIRECTORY_NAME
 
 CLIENT_ID = "AgvGDwoBRb0BJAnL2CQ8dNbE6J2fgCFIchEOyr5S"
 config_dir = Path.home() / f".{RIPPLING_DIRECTORY_NAME}"
@@ -14,12 +14,15 @@
 def get_client_id():
     return CLIENT_ID
 
-
-def get_oauth_token_data():
+def create_base_directory_if_not_exists():
     # Create the directory if it doesn't exist
     if not os.path.exists(config_dir):
         os.makedirs(config_dir)
 
+def get_oauth_token_data():
+
+    create_base_directory_if_not_exists()
+
     token_file = config_dir / OAUTH_TOKEN_FILE_NAME
     if token_file.exists():
         with token_file.open("r") as f:
@@ -29,9 +32,7 @@ def get_oauth_token_data():
 
 def save_oauth_token(token, expires_in=3600):
 
-    # Create the directory if it doesn't exist
-    if not os.path.exists(config_dir):
-        os.makedirs(config_dir)
+    create_base_directory_if_not_exists()
 
     data = {
         "token": str(token),
@@ -40,3 +41,36 @@ def save_oauth_token(token, expires_in=3600):
     token_file = config_dir / OAUTH_TOKEN_FILE_NAME
     with token_file.open("w") as f:
         json.dump(data, f)
+
+
+def get_app_config():
+    """
+    Load the app configuration from the specified directory.
+
+    Returns:
+        dict: The app configuration data.
+    """
+    create_base_directory_if_not_exists()
+
+    config_file = config_dir / APP_CONFIG_FILE
+    if config_file.exists():
+        with config_file.open("r") as f:
+            return json.load(f)
+    return {}
+
+
+def save_app_config(app_id: str, app_name: str):
+    """
+    Save the app configuration to the specified directory.
+    Args:
+        :param app_id:
+    """
+    create_base_directory_if_not_exists()
+
+    app_config = get_app_config()
+
+    app_config.update({f"{os.getcwd()}": {"id": app_id, "displayName": app_name}})
+
+    config_file = config_dir / APP_CONFIG_FILE
+    with config_file.open("w") as f:
+        json.dump(app_config, f)

rippling_cli/constants.py

@@ -1,5 +1,6 @@
 RIPPLING_DIRECTORY_NAME = "rippling_cli"
 OAUTH_TOKEN_FILE_NAME = "oauth_token.json"
+APP_CONFIG_FILE = "app_config.json"
 CODE_CHALLENGE_METHOD = "S256"
 RIPPLING_BASE_URL = "https://app.rippling.com"
 RIPPLING_API = "https://app.rippling.com/api"

rippling_cli/core/api_client.py

@@ -0,0 +1,71 @@
+import requests  # type: ignore
+
+
+class APIClient:
+    def __init__(self, base_url, headers=None):
+        self.base_url = base_url
+        self.headers = headers or {}
+
+    def make_request(self, method, endpoint, params=None, data=None):
+        url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
+        response = requests.request(method, url, params=params, json=data, headers=self.headers)
+        return response
+
+    def get(self, endpoint, params=None):
+        return self.make_request("GET", endpoint, params=params)
+
+    def post(self, endpoint, data):
+        return self.make_request("POST", endpoint, data=data)
+
+    def put(self, endpoint, data):
+        return self.make_request("PUT", endpoint, data=data)
+
+    def delete(self, endpoint, params=None):
+        return self.make_request("DELETE", endpoint, params=params)
+
+    def find_paginated(self, endpoint, page=1, page_size=10, read_preference="SECONDARY_PREFERRED"):
+        """
+        Fetch paginated data from the API.
+
+        Args:
+            endpoint (str): The API endpoint.
+            headers (dict): The headers for the API request.
+            page (int): The page number to fetch.
+            per_page (int): The number of items to fetch per page.
+
+        Yields:
+            dict: The data from the API response.
+            :param endpoint:
+            :param page:
+            :param page_size:
+            :param read_preference:
+        """
+        has_more = True
+        cursor = None
+        while has_more:
+            payload = {
+                "paginationParams": {
+                    "page": page,
+                    "cursor": cursor,
+                    "sortingMetadata": {
+                        "order": "DESC",
+                        "column": {
+                            "sortKey": "createdAt"
+                        }
+                    }
+                },
+                "pageSize": page_size,
+                "readPreference": read_preference
+            }
+            response = self.make_request("POST", f"{endpoint}/find_paginated", data=payload)
+
+            if response.status_code == 200:
+                data = response.json()
+                cursor = data.get("cursor")
+                has_more = False if not cursor else True
+                items = data["data"]
+                page += 1
+                yield items
+            else:
+                response.raise_for_status()
+                break

rippling_cli/utils/login_utils.py

@@ -0,0 +1,10 @@
+import click
+
+from rippling_cli.cli.commands.login import login
+from rippling_cli.core.oauth_token import OAuthToken
+
+
+def ensure_logged_in(ctx: click.Context):
+    if OAuthToken.is_token_expired():
+        click.echo("You are not logged in. Please log in first.")
+        ctx.invoke(login)