def fetch_all_items(func, *args, **kwargs):
all_items = []
result = func(*args, **kwargs)
while result:
all_items.extend(result)
result = func(next_page=result['next_page']) if 'next_page' in result else None
return all_items
def get_tasks_filtered_by_custom_field(workspace_id, custom_field_name, custom_field_value):
try:
projects = fetch_all_items(client.projects.get_projects, workspace=workspace_id, opt_pretty=True)
filtered_tasks = []
def process_task(task):
comments = []
for custom_field in task.get('custom_fields', []):
if custom_field['name'] == custom_field_name and custom_field.get('enum_value') and custom_field['enum_value'].get('name') == custom_field_value:
comments = fetch_all_items(client.stories.get_stories_for_task, task['gid'])
return {
'Task Name': task['name'],
'Project ID': task['gid'],
'Custom Field': custom_field['name'],
'Custom Field Value': custom_field['enum_value']['name'],
'Due Date': task.get('due_on', 'No due date'),
'Assignee': task.get('assignee', {}).get('name', 'Unassigned'),
'Created At': task.get('created_at'),
'Completed': task.get('completed'),
'Completed At': task.get('completed_at'),
'Start On': task.get('start_on'),
'Due At': task.get('due_at'),
'Notes': task.get('notes', ''),
'Num Likes': task.get('num_likes'),
'Num Hearts': task.get('num_hearts'),
'Comments': ' '.join([comment['text'] for comment in comments if comment['resource_subtype'] == 'comment']),
'Followers': ', '.join([follower['name'] for follower in task.get('followers', [])]),
'Tags': ', '.join([tag['name'] for tag in task.get('tags', [])]),
'Parent Task': task.get('parent', {}).get('name', 'No parent'),
'Workspace': task.get('workspace', {}).get('name', 'No workspace')
}
return None
with ThreadPoolExecutor(max_workers=10) as executor:
for project in projects:
tasks = fetch_all_items(client.tasks.get_tasks_for_project, project['gid'], opt_fields='name,custom_fields,due_on', opt_pretty=True)
results = executor.map(process_task, tasks)
filtered_tasks.extend([result for result in results if result])
return pd.DataFrame(filtered_tasks)
except Exception as e:
print(f"An error occurred: {e}")
return pd.DataFrame()
# Fetch filtered tasks
df_tasks = get_tasks_filtered_by_custom_field(workspace_id, 'Quarter', 'Q1')
df_tasks.to_excel('filtered_tasks.xlsx', index=False)