feat(cli): add asana cli commands

Makefile

@@ -3,6 +3,9 @@
 #
 .PHONY: install dev_install upgrade_dependencies
 
+cli_install:
+	pip install -e .
+
 install:
 	pip install -r requirements.txt
 

giges/app.py

@@ -89,6 +89,13 @@ def create_connexion_app(
     configure_logging()
     configure_sentry(flask_app)
 
+    # Import CLI packages here to avoid circular requires
+    from giges.cli.asana import (  # pylint: disable=import-outside-toplevel
+        asana_cli,
+    )
+
+    flask_app.cli.add_command(asana_cli)
+
     return connexion_app
 
 

giges/cli/asana.py

@@ -0,0 +1,85 @@
+import json
+from typing import Generator, List, Union
+
+import asana
+import click
+from flask import current_app
+from flask.cli import with_appcontext
+
+from giges.db import db
+from giges.models.asana import ResourceTypeEnum, Webhook
+
+
+def create_client() -> asana.Client:
+    return asana.Client.access_token(current_app.config["ASANA_TOKEN"])
+
+
+def print_response(response: Union[Generator, List]) -> None:
+    if not isinstance(response, dict):
+        response = list(response)
+    print(json.dumps(response, indent=4))
+
+
+@click.group(name="asana", help="Manage asana information from giges")
+def asana_cli() -> None:
+    pass
+
+
+@asana_cli.command(help="List all workspaces in Asana (spoiler, only one)")
+@with_appcontext
+def list_workspaces() -> None:
+    client = create_client()
+    print_response(current_app.config)
+    print_response(client.workspaces.get_workspaces())
+
+
+@asana_cli.command(help="Configure a webhook for the asana projects")
+@with_appcontext
+def create_projects_webhook() -> None:
+    client = create_client()
+    path = "/asana/projects"
+
+    webhook = Webhook(path=path, resource_type=ResourceTypeEnum.project)
+    db.session.add(webhook)
+    db.session.commit()
+
+    response = client.webhooks.create(
+        resource=current_app.config["ASANA_WORKSPACE"],
+        target=f"{current_app.config['SERVER_BASE_URI']}{path}",
+        filters=[{"resource_type": "project"}],
+    )
+
+    print_response(response)
+    webhook.external_id = response["gid"]
+    db.session.add(webhook)
+    db.session.commit()
+
+
+@asana_cli.command(help="List currently configured Asana webhooks")
+@with_appcontext
+def list_webhooks() -> None:
+    client = create_client()
+    print_response(
+        client.webhooks.get_webhooks(
+            workspace=current_app.config["ASANA_WORKSPACE"]
+        )
+    )
+
+
+@asana_cli.command(help="List all Asana projects")
+@with_appcontext
+def list_projects() -> None:
+    client = create_client()
+    print_response(
+        client.projects.find_all(
+            workspace=current_app.config["ASANA_WORKSPACE"]
+        )
+    )
+
+
+@asana_cli.command(help="Delete a currently configured Asana webhook")
+@click.argument("webhook_id", required=True)
+@with_appcontext
+def delete_webhook(webhook_id: str) -> None:
+    client = create_client()
+    print_response(client.webhooks.delete_by_id(webhook_id))

giges/settings.py

@@ -7,34 +7,46 @@ class BaseSettings:
     APP_VERSION = os.getenv("APP_VERSION", "latest")
 
     ENVIRONMENT = ""
-    SENTRY_URI = ""
+    SERVER_BASE_URI = ""
     DEBUG = os.getenv("GIGES_DEBUG", "0") == "1"
 
     # Database
     SQLALCHEMY_DATABASE_URI = os.getenv("GIGES_DATABASE_URI")
     SQLALCHEMY_TRACK_MODIFICATIONS = False
 
+    # Integrations
+    SENTRY_URI = ""
+    ASANA_WORKSPACE = ""
+    ASANA_TOKEN = os.environ.get("ASANA_TOKEN", "")
+
 
 class ProductionSettings(BaseSettings):
     ENVIRONMENT = "production"
 
+    ASANA_WORKSPACE = "1199978051314275"
+    SERVER_BASE_URI = "https://integrations.tesselo.com"
     SENTRY_URI = "https://[email protected]/5911249"  # noqa: E501
 
 
 class StagingSettings(BaseSettings):
     ENVIRONMENT = "staging"
 
+    ASANA_WORKSPACE = "1200469161331386"
+    SERVER_BASE_URI = "https://integrations-staging.tesselo.com"
+
 
 class DevelopmentSettings(BaseSettings):
     ENVIRONMENT = "development"
     DEBUG = True
+    SERVER_BASE_URI = "http://localhost:8080"
     SQLALCHEMY_DATABASE_URI = (
         f"sqlite:///{Path(__file__).parents[1]}/development.db"
     )
 
 
 class TestingSettings(BaseSettings):
     ENVIRONMENT = "testing"
+    SERVER_BASE_URI = "http://localhost:8080"
     SQLALCHEMY_DATABASE_URI = (
         f"sqlite:///{Path(__file__).parents[1]}/testing.db"
     )