Build a Plane App (Beta)

import os
from urllib.parse import urlencode

params = {
    "client_id": os.getenv("PLANE_CLIENT_ID"),
    "response_type": "code",
    "redirect_uri": os.getenv("PLANE_REDIRECT_URI"),
    # Optional: include state if needed
}

consent_url = f"https://api.plane.so/auth/o/authorize-app/?{urlencode(params)}"

import os
import time
from plane.oauth.api import OAuthApi
from plane.oauth.models import OAuthConfig

# Initialize OAuth API
def get_oauth_api():
    oauth_config = OAuthConfig(
        client_id=os.getenv("PLANE_CLIENT_ID"),
        client_secret=os.getenv("PLANE_CLIENT_SECRET"),
        redirect_uri=os.getenv("PLANE_REDIRECT_URI"),
    )
    return OAuthApi(
        oauth_config=oauth_config,
        base_url=os.getenv("PLANE_BASE_URL", "https://api.plane.so"),
    )

# Get bot token using app installation ID
oauth_api = get_oauth_api()
token_response = oauth_api.get_bot_token(app_installation_id)

# Get app installation details
app_installations = oauth_api.get_app_installations(
    token_response.access_token,
    app_installation_id,
)

if not app_installations:
    raise Exception(f"No app installations found for app installation ID {app_installation_id}")

app_installation = app_installations[0]
bot_token = token_response.access_token
expires_in = token_response.expires_in

from plane.oauth.api import OAuthApi
from plane.oauth.models import OAuthConfig

# Initialize OAuth API
oauth_api = get_oauth_api()  # Using the helper function from above

# Exchange authorization code for access and refresh tokens
code = "authorization_code_from_callback"
token_response = oauth_api.exchange_code_for_token(
    code,
    "authorization_code",
)

# Parse the response
access_token = token_response.access_token
refresh_token = token_response.refresh_token
expires_in = token_response.expires_in

# Using the OAuth API to fetch app installation details
oauth_api = get_oauth_api()
app_installations = oauth_api.get_app_installations(
    token,  # Either access token or bot token
    app_installation_id,
)

if app_installations:
    workspace_details = app_installations[0]
    print(f"Workspace: {workspace_details.workspace_detail.name}")
    print(f"Workspace Slug: {workspace_details.workspace_detail.slug}")
    print(f"Bot User ID: {workspace_details.app_bot}")

interface WebhookPayload {
  event: string;
  action: string;
  webhook_id: string;
  workspace_id: string;
  data: any;
  activity: {
    actor: {
      id: string;
      display_name: string;
      email?: string;
    };
    field?: string;
    new_value?: any;
    old_value?: any;
  };
}

// Process incoming webhook
async function handleWebhook(payload: WebhookPayload) {
  console.log(`Received ${payload.event} ${payload.action} event`);
  
  // Get workspace credentials
  const credentials = await getCredentialsForWorkspace(payload.workspace_id);
  if (!credentials) {
    throw new Error(`No credentials found for workspace ${payload.workspace_id}`);
  }
  
  // Process specific event types
  if (payload.event === 'issue_comment' && payload.action === 'created') {
    const comment = payload.data.comment_stripped;
    if (comment.includes('/your-command')) {
      // Handle your custom command
      await processCommand(payload.data, credentials);
    }
  }
}

# When bot token expires, request a new one using the same app_installation_id
from plane.oauth.api import OAuthApi

def refresh_bot_token(app_installation_id: str):
    """Refresh an expired bot token"""
    oauth_api = get_oauth_api()  # Using helper function from earlier examples
    
    # Get new bot token using the same app_installation_id
    token_response = oauth_api.get_bot_token(app_installation_id)
    
    # Store the new token securely in your database
    new_bot_token = token_response.access_token
    expires_in = token_response.expires_in
    
    return new_bot_token, expires_in

# Usage example
new_token, expires_in = refresh_bot_token(app_installation_id)

# When access token expires, use refresh token to get a new access token
from plane.oauth.api import OAuthApi

def refresh_user_token(refresh_token: str):
    """Refresh an expired user access token"""
    oauth_api = get_oauth_api()  # Using helper function from earlier examples
    
    # Use refresh token to get new access token
    token_response = oauth_api.exchange_code_for_token(
        refresh_token,
        "refresh_token",
    )
    
    # Store the new tokens securely
    new_access_token = token_response.access_token
    new_refresh_token = token_response.refresh_token  # May be the same or new
    expires_in = token_response.expires_in
    
    return new_access_token, new_refresh_token, expires_in

# Usage example
new_access_token, new_refresh_token, expires_in = refresh_user_token(stored_refresh_token)