How to Revenue Report Asana: Expert Tips

import requests
import time
import logging
import os
import json
from typing import Dict, List, Optional, Any
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure module-level logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class AsanaAPIClient:
    '''Production-grade Asana API v3 client with rate limiting, retry logic, and pagination support.'''

    BASE_URL = 'https://app.asana.com/api/v3'
    MAX_RETRIES = 5
    RATE_LIMIT_SLEEP = 60  # Seconds to wait on 429 responses

    def __init__(self, access_token: str, timeout: int = 30):
        '''Initialize client with Asana personal access token.

        Args:
            access_token: Asana PAT with `projects:read`, `tasks:read`, `custom_fields:read` scopes
            timeout: Request timeout in seconds (default 30)
        '''
        self.access_token = access_token
        self.timeout = timeout
        self.session = self._configure_session()

    def _configure_session(self) -> requests.Session:
        '''Configure requests session with retry logic for transient errors.'''
        session = requests.Session()
        retry_strategy = Retry(
            total=self.MAX_RETRIES,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=['GET', 'POST'],
            backoff_factor=1  # Exponential backoff: 1, 2, 4, 8... seconds
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount('https://', adapter)
        session.headers.update({
            'Authorization': f'Bearer {self.access_token}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        })
        return session

    def _handle_rate_limit(self, response: requests.Response) -> None:
        '''Handle 429 Rate Limit Exceeded responses per Asana API docs.'''
        if response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', self.RATE_LIMIT_SLEEP))
            logger.warning(f'Rate limited. Sleeping for {retry_after} seconds')
            time.sleep(retry_after)
            response.raise_for_status()  # Retry after sleep

    def get_projects(self, team_gid: str, archived: bool = False) -> List[Dict[str, Any]]:
        '''Fetch all projects for a given Asana team, handling pagination.

        Args:
            team_gid: Asana team GID (globally unique identifier)
            archived: Whether to include archived projects (default False)

        Returns:
            List of project dicts with gid, name, created_at, custom_field_settings
        '''
        projects = []
        url = f'{self.BASE_URL}/teams/{team_gid}/projects'
        params = {
            'archived': 'true' if archived else 'false',
            'opt_fields': 'gid,name,created_at,modified_at,custom_field_settings,currency_code,public'
        }

        while url:
            try:
                response = self.session.get(url, params=params, timeout=self.timeout)
                self._handle_rate_limit(response)
                response.raise_for_status()
                data = response.json()

                # Extract projects from response
                projects.extend(data.get('data', []))

                # Handle pagination: Asana returns next page URL in 'next' field
                next_page = data.get('next')
                url = next_page.get('path') if next_page else None
                params = None  # Parameters are included in next page URL

                logger.info(f'Fetched {len(projects)} total projects so far')

            except requests.exceptions.HTTPError as e:
                if e.response.status_code == 401:
                    logger.error('Invalid Asana access token. Check PAT scopes.')
                    raise
                elif e.response.status_code == 404:
                    logger.error(f'Team {team_gid} not found. Verify team GID.')
                    raise
                else:
                    logger.error(f'HTTP error fetching projects: {e}')
                    raise
            except requests.exceptions.RequestException as e:
                logger.error(f'Network error fetching projects: {e}')
                raise

        return projects

if __name__ == '__main__':
    # Example usage: Fetch projects for a team (replace with your team GID)
    ASANA_PAT = os.getenv('ASANA_ACCESS_TOKEN')
    if not ASANA_PAT:
        raise ValueError('Set ASANA_ACCESS_TOKEN environment variable')

    client = AsanaAPIClient(access_token=ASANA_PAT)
    try:
        projects = client.get_projects(team_gid='1234567890123456')  # Replace with real team GID
        print(f'Fetched {len(projects)} projects')
        with open('projects.json', 'w') as f:
            json.dump(projects, f, indent=2)
    except Exception as e:
        logger.error(f'Failed to fetch projects: {e}')

import pandas as pd
import json
import logging
import os
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import pytz

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AsanaRevenueCalculator:
    '''Calculate attributed revenue from Asana project and task data with multi-currency support.'''

    # Asana custom field GIDs (replace with your own from Asana settings)
    DEAL_SIZE_FIELD = '1234567890123456'  # Custom field: Deal Size (currency)
    CLOSE_DATE_FIELD = '2345678901234567'  # Custom field: Close Date (date)
    RENEWAL_STATUS_FIELD = '3456789012345678'  # Custom field: Renewal Status (enum)
    CURRENCY_FIELD = '4567890123456789'  # Custom field: Currency Code (enum)

    # Exchange rates (USD base, update daily via API in production)
    EXCHANGE_RATES = {
        'USD': 1.0,
        'EUR': 1.08,
        'GBP': 1.27,
        'JPY': 0.0067,
        'CAD': 0.74
    }

    def __init__(self, data_path: str, timezone: str = 'UTC'):
        '''Initialize calculator with path to Asana data JSON.

        Args:
            data_path: Path to projects.json or tasks.json exported from Asana API
            timezone: Timezone for date calculations (default UTC)
        '''
        self.data_path = data_path
        self.timezone = pytz.timezone(timezone)
        self.raw_data = self._load_data()
        self.df = self._normalize_to_dataframe()

    def _load_data(self) -> List[Dict]:
        '''Load and validate raw Asana data from JSON file.'''
        if not os.path.exists(self.data_path):
            raise FileNotFoundError(f'Data file not found: {self.data_path}')

        try:
            with open(self.data_path, 'r') as f:
                data = json.load(f)
            logger.info(f'Loaded {len(data)} records from {self.data_path}')
            return data
        except json.JSONDecodeError as e:
            logger.error(f'Invalid JSON in {self.data_path}: {e}')
            raise

    def _normalize_to_dataframe(self) -> pd.DataFrame:
        '''Normalize nested Asana project data to flat Pandas DataFrame.'''
        rows = []
        for project in self.raw_data:
            # Extract base project fields
            row = {
                'project_gid': project.get('gid'),
                'project_name': project.get('name'),
                'created_at': project.get('created_at'),
                'modified_at': project.get('modified_at'),
                'currency_code': project.get('currency_code', 'USD')
            }

            # Extract custom field values (flatten nested custom field settings)
            custom_fields = {
                cf['custom_field']['gid']: cf['value']
                for cf in project.get('custom_field_settings', [])
                if cf.get('value') is not None
            }

            # Map custom fields to readable names
            row['deal_size'] = custom_fields.get(self.DEAL_SIZE_FIELD, 0.0)
            row['close_date'] = custom_fields.get(self.CLOSE_DATE_FIELD)
            row['renewal_status'] = custom_fields.get(self.RENEWAL_STATUS_FIELD, 'new')
            row['project_currency'] = custom_fields.get(self.CURRENCY_FIELD, row['currency_code'])

            # Skip projects with no deal size (non-revenue generating)
            if row['deal_size'] == 0.0:
                continue

            rows.append(row)

        df = pd.DataFrame(rows)

        # Convert date columns to datetime
        df['close_date'] = pd.to_datetime(df['close_date'], errors='coerce')
        df['created_at'] = pd.to_datetime(df['created_at'], errors='coerce')
        df['modified_at'] = pd.to_datetime(df['modified_at'], errors='coerce')

        # Filter out rows with invalid close dates
        df = df.dropna(subset=['close_date'])
        logger.info(f'Normalized to {len(df)} revenue-generating projects')

        return df

    def calculate_period_revenue(self, start_date: datetime, end_date: datetime, currency: str = 'USD') -> Dict:
        '''Calculate total attributed revenue for a given date range.

        Args:
            start_date: Start of reporting period (timezone-aware)
            end_date: End of reporting period (timezone-aware)
            currency: Target currency for conversion (default USD)

        Returns:
            Dict with total_revenue, new_revenue, renewal_revenue, project_count
        '''
        # Filter projects by close date
        mask = (self.df['close_date'] >= start_date) & (self.df['close_date'] <= end_date)
        period_df = self.df[mask].copy()

        if len(period_df) == 0:
            return {
                'total_revenue': 0.0,
                'new_revenue': 0.0,
                'renewal_revenue': 0.0,
                'project_count': 0,
                'currency': currency
            }

        # Convert all deal sizes to target currency
        period_df['exchange_rate'] = period_df['project_currency'].map(self.EXCHANGE_RATES)
        period_df['exchange_rate'] = period_df['exchange_rate'].fillna(1.0)  # Default to USD if currency not found
        period_df['deal_size_usd'] = period_df['deal_size'] * period_df['exchange_rate']

        # Calculate target currency conversion
        target_rate = self.EXCHANGE_RATES.get(currency, 1.0)
        period_df['deal_size_target'] = period_df['deal_size_usd'] * target_rate

        # Split by renewal status
        new_df = period_df[period_df['renewal_status'] == 'new']
        renewal_df = period_df[period_df['renewal_status'] == 'renewal']

        return {
            'total_revenue': round(period_df['deal_size_target'].sum(), 2),
            'new_revenue': round(new_df['deal_size_target'].sum(), 2),
            'renewal_revenue': round(renewal_df['deal_size_target'].sum(), 2),
            'project_count': len(period_df),
            'currency': currency
        }

if __name__ == '__main__':
    # Example usage: Calculate Q3 2024 revenue
    calculator = AsanaRevenueCalculator(data_path='projects.json')

    start = datetime(2024, 7, 1, tzinfo=pytz.UTC)
    end = datetime(2024, 9, 30, tzinfo=pytz.UTC)

    q3_revenue = calculator.calculate_period_revenue(start, end, currency='USD')
    print(f'Q3 2024 Revenue: ${q3_revenue["total_revenue"]} USD')
    print(f'New Deals: ${q3_revenue["new_revenue"]}, Renewals: ${q3_revenue["renewal_revenue"]}')
    print(f'Total Projects: {q3_revenue["project_count"]}')

import os
import logging
import pandas as pd
import json
from typing import Dict
from sqlalchemy import create_engine, text
from great_expectations.dataset import PandasDataset
import requests

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RevenueReportPipeline:
    '''End-to-end pipeline to validate, store, and distribute Asana revenue reports.'''

    def __init__(self, db_url: str, slack_webhook: Optional[str] = None):
        '''Initialize pipeline with database and optional Slack webhook.

        Args:
            db_url: SQLAlchemy-compatible database URL (e.g., postgresql://user:pass@localhost:5432/revenue)
            slack_webhook: Slack incoming webhook URL for report notifications
        '''
        self.db_url = db_url
        self.slack_webhook = slack_webhook
        self.engine = create_engine(db_url)

    def validate_data(self, df: pd.DataFrame) -> bool:
        '''Validate revenue data with Great Expectations to catch anomalies.

        Args:
            df: Pandas DataFrame of revenue projects

        Returns:
            True if data passes all expectations, False otherwise
        '''
        try:
            # Wrap DataFrame in Great Expectations dataset
            ge_df = PandasDataset(df)

            # Expect non-null project GIDs
            result = ge_df.expect_column_values_to_not_be_null('project_gid')
            if not result['success']:
                logger.error('Validation failed: Null project GIDs found')
                return False

            # Expect deal size to be positive
            result = ge_df.expect_column_values_to_be_between('deal_size', min_value=0.01)
            if not result['success']:
                logger.error('Validation failed: Deal size <= 0 found')
                return False

            # Expect close date to be in the past (no future-dated deals)
            today = pd.Timestamp.now(tz='UTC')
            result = ge_df.expect_column_values_to_be_between(
                'close_date', 
                min_value=pd.Timestamp('2020-01-01', tz='UTC'), 
                max_value=today
            )
            if not result['success']:
                logger.error('Validation failed: Future close dates found')
                return False

            # Expect currency codes to be in supported list
            supported_currencies = ['USD', 'EUR', 'GBP', 'JPY', 'CAD']
            result = ge_df.expect_column_values_to_be_in_set('project_currency', supported_currencies)
            if not result['success']:
                logger.error('Validation failed: Unsupported currency codes found')
                return False

            logger.info('All data validation checks passed')
            return True

        except Exception as e:
            logger.error(f'Validation error: {e}')
            return False

    def store_reports(self, daily: Dict, weekly: Dict, monthly: Dict) -> None:
        '''Store report summaries in PostgreSQL for historical tracking.

        Args:
            daily: Daily revenue dict from calculator
            weekly: Weekly revenue dict
            monthly: Monthly revenue dict
        '''
        with self.engine.begin() as conn:
            # Create table if not exists
            conn.execute(text('''
                CREATE TABLE IF NOT EXISTS asana_revenue_reports (
                    id SERIAL PRIMARY KEY,
                    report_date DATE NOT NULL,
                    period_type VARCHAR(10) NOT NULL,
                    total_revenue NUMERIC(12,2) NOT NULL,
                    new_revenue NUMERIC(12,2) NOT NULL,
                    renewal_revenue NUMERIC(12,2) NOT NULL,
                    project_count INT NOT NULL,
                    currency VARCHAR(3) NOT NULL,
                    created_at TIMESTAMP DEFAULT NOW()
                )
            '''))

            # Insert daily report
            conn.execute(text('''
                INSERT INTO asana_revenue_reports (report_date, period_type, total_revenue, new_revenue, renewal_revenue, project_count, currency)
                VALUES (:report_date, 'daily', :total, :new, :renewal, :count, :currency)
            '''), {
                'report_date': pd.Timestamp.now(tz='UTC').date(),
                'total': daily['total_revenue'],
                'new': daily['new_revenue'],
                'renewal': daily['renewal_revenue'],
                'count': daily['project_count'],
                'currency': daily['currency']
            })

            # Insert weekly report
            conn.execute(text('''
                INSERT INTO asana_revenue_reports (report_date, period_type, total_revenue, new_revenue, renewal_revenue, project_count, currency)
                VALUES (:report_date, 'weekly', :total, :new, :renewal, :count, :currency)
            '''), {
                'report_date': pd.Timestamp.now(tz='UTC').date(),
                'total': weekly['total_revenue'],
                'new': weekly['new_revenue'],
                'renewal': weekly['renewal_revenue'],
                'count': weekly['project_count'],
                'currency': weekly['currency']
            })

            # Insert monthly report
            conn.execute(text('''
                INSERT INTO asana_revenue_reports (report_date, period_type, total_revenue, new_revenue, renewal_revenue, project_count, currency)
                VALUES (:report_date, 'monthly', :total, :new, :renewal, :count, :currency)
            '''), {
                'report_date': pd.Timestamp.now(tz='UTC').date(),
                'total': monthly['total_revenue'],
                'new': monthly['new_revenue'],
                'renewal': monthly['renewal_revenue'],
                'count': monthly['project_count'],
                'currency': monthly['currency']
            })

            logger.info('Stored daily, weekly, and monthly reports in PostgreSQL')

    def send_slack_notification(self, daily: Dict) -> None:
        '''Send daily revenue summary to Slack via webhook.'''
        if not self.slack_webhook:
            logger.warning('No Slack webhook configured, skipping notification')
            return

        message = {
            'text': f'📊 *Asana Daily Revenue Report* ({daily["currency"]})',
            'blocks': [
                {
                    'type': 'section',
                    'text': {
                        'type': 'mrkdwn',
                        'text': f'*Total Revenue Today:* ${daily["total_revenue"]}\n*New Deals:* ${daily["new_revenue"]}\n*Renewals:* ${daily["renewal_revenue"]}\n*Projects Closed:* {daily["project_count"]}'
                    }
                }
            ]
        }

        try:
            response = requests.post(self.slack_webhook, json=message, timeout=10)
            response.raise_for_status()
            logger.info('Slack notification sent successfully')
        except Exception as e:
            logger.error(f'Failed to send Slack notification: {e}')

if __name__ == '__main__':
    # Example usage: Run full pipeline
    DB_URL = os.getenv('REVENUE_DB_URL', 'postgresql://localhost/revenue')
    SLACK_WEBHOOK = os.getenv('SLACK_WEBHOOK_URL')

    # Load pre-calculated data (from previous step)
    from revenue_calculator import AsanaRevenueCalculator
    calculator = AsanaRevenueCalculator('projects.json')
    now = pd.Timestamp.now(tz='UTC')

    daily = calculator.calculate_period_revenue(
        start_date=now - pd.Timedelta(days=1),
        end_date=now,
        currency='USD'
    )
    weekly = calculator.calculate_period_revenue(
        start_date=now - pd.Timedelta(days=7),
        end_date=now,
        currency='USD'
    )
    monthly = calculator.calculate_period_revenue(
        start_date=now - pd.Timedelta(days=30),
        end_date=now,
        currency='USD'
    )

    # Run pipeline
    pipeline = RevenueReportPipeline(db_url=DB_URL, slack_webhook=SLACK_WEBHOOK)

    # Validate data (load df from calculator)
    if pipeline.validate_data(calculator.df):
        pipeline.store_reports(daily, weekly, monthly)
        pipeline.send_slack_notification(daily)
    else:
        logger.error('Data validation failed, aborting report generation')

def get_projects_batch(self, team_gids: List[str]) -> List[Dict]:
    batch_payload = [{'method': 'GET', 'relative_url': f'/teams/{gid}/projects?opt_fields=gid,name'} for gid in team_gids]
    response = self.session.post(f'{self.BASE_URL}/batch', json={'data': batch_payload})
    # Handle partial failures per batch element
    projects = []
    for i, res in enumerate(response.json()['data']):
        if res['status'] == 200:
            projects.extend(res['body']['data'])
        else:
            logger.warning(f'Batch request for team {team_gids[i]} failed: {res["status"]}')
    return projects

# GitHub Actions step to validate custom field GIDs
- name: Validate Asana Custom Field GIDs
  run: |
    python -c "
    import os, requests
    ASANA_PAT = os.getenv('ASANA_ACCESS_TOKEN')
    headers = {'Authorization': f'Bearer {ASANA_PAT}'}
    fields = {
        'DEAL_SIZE': ('1234567890123456', 'Deal Size'),
        'CLOSE_DATE': ('2345678901234567', 'Close Date')
    }
    for key, (gid, expected_name) in fields.items():
        res = requests.get(f'https://app.asana.com/api/v3/custom_fields/{gid}', headers=headers)
        if res.json()['data']['name'] != expected_name:
            raise ValueError(f'Custom field {key} GID mismatch: expected {expected_name}, got {res.json()["data"]["name"]}')
    "

# Write report DataFrame to PostgreSQL
from sqlalchemy import create_engine
engine = create_engine('postgresql://user:pass@managed-db.example.com/revenue')
report_df.to_sql(
    name='asana_revenue_reports',
    con=engine,
    if_exists='append',
    index=False,
    dtype={
        'report_date': 'DATE',
        'total_revenue': 'NUMERIC(12,2)',
        'project_count': 'INT'
    }
)

asana-revenue-reporting/
├── README.md                # Setup instructions, env vars, deploy steps
├── requirements.txt         # Python dependencies (requests, pandas, great-expectations, sqlalchemy)
├── src/
│   ├── __init__.py
│   ├── asana_client.py      # Asana API client (Code Example 1)
│   ├── revenue_calculator.py # Revenue calculation engine (Code Example 2)
│   └── report_pipeline.py   # Validation, storage, Slack alerts (Code Example 3)
├── config/
│   ├── custom_fields.json   # Custom field GID mappings
│   └── exchange_rates.json  # Daily exchange rate cache
├── tests/
│   ├── test_asana_client.py # Unit tests for API client
│   └── test_calculator.py   # Unit tests for revenue calculator
└── .github/
    └── workflows/
        └── validate.yml      # Deploy-time custom field validation (Tip 2)