c3c96712e6
Updated the get_todos method to retrieve all todos from a todolist by handling pagination transparently. The method now aggregates todos across multiple pages and updates the README to reflect this change.
900 lines
36 KiB
Python
900 lines
36 KiB
Python
import os
|
|
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
class BasecampClient:
|
|
"""
|
|
Client for interacting with Basecamp 3 API using Basic Authentication or OAuth 2.0.
|
|
"""
|
|
|
|
def __init__(self, username=None, password=None, account_id=None, user_agent=None,
|
|
access_token=None, auth_mode="basic"):
|
|
"""
|
|
Initialize the Basecamp client with credentials.
|
|
|
|
Args:
|
|
username (str, optional): Basecamp username (email) for Basic Auth
|
|
password (str, optional): Basecamp password for Basic Auth
|
|
account_id (str, optional): Basecamp account ID
|
|
user_agent (str, optional): User agent for API requests
|
|
access_token (str, optional): OAuth access token for OAuth Auth
|
|
auth_mode (str, optional): Authentication mode ('basic' or 'oauth')
|
|
"""
|
|
# Load environment variables if not provided directly
|
|
load_dotenv()
|
|
|
|
self.auth_mode = auth_mode.lower()
|
|
self.account_id = account_id or os.getenv('BASECAMP_ACCOUNT_ID')
|
|
self.user_agent = user_agent or os.getenv('USER_AGENT')
|
|
|
|
# Set up authentication based on mode
|
|
if self.auth_mode == 'basic':
|
|
self.username = username or os.getenv('BASECAMP_USERNAME')
|
|
self.password = password or os.getenv('BASECAMP_PASSWORD')
|
|
|
|
if not all([self.username, self.password, self.account_id, self.user_agent]):
|
|
raise ValueError("Missing required credentials for Basic Auth. Set them in .env file or pass them to the constructor.")
|
|
|
|
self.auth = (self.username, self.password)
|
|
self.headers = {
|
|
"User-Agent": self.user_agent,
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
elif self.auth_mode == 'oauth':
|
|
self.access_token = access_token or os.getenv('BASECAMP_ACCESS_TOKEN')
|
|
|
|
if not all([self.access_token, self.account_id, self.user_agent]):
|
|
raise ValueError("Missing required credentials for OAuth. Set them in .env file or pass them to the constructor.")
|
|
|
|
self.auth = None # No basic auth needed for OAuth
|
|
self.headers = {
|
|
"User-Agent": self.user_agent,
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self.access_token}"
|
|
}
|
|
|
|
else:
|
|
raise ValueError("Invalid auth_mode. Must be 'basic' or 'oauth'")
|
|
|
|
# Basecamp 3 uses a different URL structure
|
|
self.base_url = f"https://3.basecampapi.com/{self.account_id}"
|
|
|
|
def test_connection(self):
|
|
"""Test the connection to Basecamp API."""
|
|
response = self.get('projects.json')
|
|
if response.status_code == 200:
|
|
return True, "Connection successful"
|
|
else:
|
|
return False, f"Connection failed: {response.status_code} - {response.text}"
|
|
|
|
def get(self, endpoint, params=None):
|
|
"""Make a GET request to the Basecamp API."""
|
|
url = f"{self.base_url}/{endpoint}"
|
|
return requests.get(url, auth=self.auth, headers=self.headers, params=params)
|
|
|
|
def post(self, endpoint, data=None):
|
|
"""Make a POST request to the Basecamp API."""
|
|
url = f"{self.base_url}/{endpoint}"
|
|
return requests.post(url, auth=self.auth, headers=self.headers, json=data)
|
|
|
|
def put(self, endpoint, data=None):
|
|
"""Make a PUT request to the Basecamp API."""
|
|
url = f"{self.base_url}/{endpoint}"
|
|
return requests.put(url, auth=self.auth, headers=self.headers, json=data)
|
|
|
|
def delete(self, endpoint):
|
|
"""Make a DELETE request to the Basecamp API."""
|
|
url = f"{self.base_url}/{endpoint}"
|
|
return requests.delete(url, auth=self.auth, headers=self.headers)
|
|
|
|
def patch(self, endpoint, data=None):
|
|
"""Make a PATCH request to the Basecamp API."""
|
|
url = f"{self.base_url}/{endpoint}"
|
|
return requests.patch(url, auth=self.auth, headers=self.headers, json=data)
|
|
|
|
# Project methods
|
|
def get_projects(self):
|
|
"""Get all projects."""
|
|
response = self.get('projects.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get projects: {response.status_code} - {response.text}")
|
|
|
|
def get_project(self, project_id):
|
|
"""Get a specific project by ID."""
|
|
response = self.get(f'projects/{project_id}.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get project: {response.status_code} - {response.text}")
|
|
|
|
# To-do list methods
|
|
def get_todoset(self, project_id):
|
|
"""Get the todoset for a project (Basecamp 3 has one todoset per project)."""
|
|
project = self.get_project(project_id)
|
|
try:
|
|
return next(_ for _ in project["dock"] if _["name"] == "todoset")
|
|
except (IndexError, TypeError, StopIteration):
|
|
raise Exception(f"Failed to get todoset for project: {project.id}. Project response: {project}")
|
|
|
|
def get_todolists(self, project_id):
|
|
"""Get all todolists for a project."""
|
|
# First get the todoset ID for this project
|
|
todoset = self.get_todoset(project_id)
|
|
todoset_id = todoset['id']
|
|
|
|
# Then get all todolists in this todoset
|
|
response = self.get(f'buckets/{project_id}/todosets/{todoset_id}/todolists.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get todolists: {response.status_code} - {response.text}")
|
|
|
|
def get_todolist(self, todolist_id):
|
|
"""Get a specific todolist."""
|
|
response = self.get(f'todolists/{todolist_id}.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get todolist: {response.status_code} - {response.text}")
|
|
|
|
# To-do methods
|
|
def get_todos(self, project_id, todolist_id):
|
|
"""Get all todos in a todolist, handling pagination.
|
|
|
|
Basecamp paginates list endpoints (commonly 15 items per page). This
|
|
implementation follows pagination via the `page` query parameter and
|
|
the HTTP `Link` header if present, aggregating all pages before
|
|
returning the combined list.
|
|
"""
|
|
endpoint = f'buckets/{project_id}/todolists/{todolist_id}/todos.json'
|
|
|
|
all_todos = []
|
|
page = 1
|
|
|
|
while True:
|
|
response = self.get(endpoint, params={"page": page})
|
|
if response.status_code != 200:
|
|
raise Exception(f"Failed to get todos: {response.status_code} - {response.text}")
|
|
|
|
page_items = response.json() or []
|
|
all_todos.extend(page_items)
|
|
|
|
# Check for next page using Link header or by empty result
|
|
link_header = response.headers.get("Link", "")
|
|
has_next = 'rel="next"' in link_header if link_header else False
|
|
|
|
if not page_items or not has_next:
|
|
break
|
|
|
|
page += 1
|
|
|
|
return all_todos
|
|
|
|
def get_todo(self, todo_id):
|
|
"""Get a specific todo."""
|
|
response = self.get(f'todos/{todo_id}.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get todo: {response.status_code} - {response.text}")
|
|
|
|
def create_todo(self, project_id, todolist_id, content, description=None, assignee_ids=None,
|
|
completion_subscriber_ids=None, notify=False, due_on=None, starts_on=None):
|
|
"""
|
|
Create a new todo item in a todolist.
|
|
|
|
Args:
|
|
project_id (str): Project ID
|
|
todolist_id (str): Todolist ID
|
|
content (str): The todo item's text (required)
|
|
description (str, optional): HTML description
|
|
assignee_ids (list, optional): List of person IDs to assign
|
|
completion_subscriber_ids (list, optional): List of person IDs to notify on completion
|
|
notify (bool, optional): Whether to notify assignees
|
|
due_on (str, optional): Due date in YYYY-MM-DD format
|
|
starts_on (str, optional): Start date in YYYY-MM-DD format
|
|
|
|
Returns:
|
|
dict: The created todo
|
|
"""
|
|
endpoint = f'buckets/{project_id}/todolists/{todolist_id}/todos.json'
|
|
data = {'content': content}
|
|
|
|
if description is not None:
|
|
data['description'] = description
|
|
if assignee_ids is not None:
|
|
data['assignee_ids'] = assignee_ids
|
|
if completion_subscriber_ids is not None:
|
|
data['completion_subscriber_ids'] = completion_subscriber_ids
|
|
if notify is not None:
|
|
data['notify'] = notify
|
|
if due_on is not None:
|
|
data['due_on'] = due_on
|
|
if starts_on is not None:
|
|
data['starts_on'] = starts_on
|
|
|
|
response = self.post(endpoint, data)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to create todo: {response.status_code} - {response.text}")
|
|
|
|
def update_todo(self, project_id, todo_id, content=None, description=None, assignee_ids=None,
|
|
completion_subscriber_ids=None, notify=None, due_on=None, starts_on=None):
|
|
"""
|
|
Update an existing todo item.
|
|
|
|
Args:
|
|
project_id (str): Project ID
|
|
todo_id (str): Todo ID
|
|
content (str, optional): The todo item's text
|
|
description (str, optional): HTML description
|
|
assignee_ids (list, optional): List of person IDs to assign
|
|
completion_subscriber_ids (list, optional): List of person IDs to notify on completion
|
|
notify (bool, optional): Whether to notify assignees
|
|
due_on (str, optional): Due date in YYYY-MM-DD format
|
|
starts_on (str, optional): Start date in YYYY-MM-DD format
|
|
|
|
Returns:
|
|
dict: The updated todo
|
|
"""
|
|
endpoint = f'buckets/{project_id}/todos/{todo_id}.json'
|
|
data = {}
|
|
|
|
if content is not None:
|
|
data['content'] = content
|
|
if description is not None:
|
|
data['description'] = description
|
|
if assignee_ids is not None:
|
|
data['assignee_ids'] = assignee_ids
|
|
if completion_subscriber_ids is not None:
|
|
data['completion_subscriber_ids'] = completion_subscriber_ids
|
|
if notify is not None:
|
|
data['notify'] = notify
|
|
if due_on is not None:
|
|
data['due_on'] = due_on
|
|
if starts_on is not None:
|
|
data['starts_on'] = starts_on
|
|
|
|
if not data:
|
|
raise ValueError("No fields provided to update")
|
|
|
|
response = self.put(endpoint, data)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to update todo: {response.status_code} - {response.text}")
|
|
|
|
def delete_todo(self, project_id, todo_id):
|
|
"""
|
|
Delete a todo item.
|
|
|
|
Args:
|
|
project_id (str): Project ID
|
|
todo_id (str): Todo ID
|
|
|
|
Returns:
|
|
bool: True if successful
|
|
"""
|
|
endpoint = f'buckets/{project_id}/todos/{todo_id}.json'
|
|
response = self.delete(endpoint)
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to delete todo: {response.status_code} - {response.text}")
|
|
|
|
def complete_todo(self, project_id, todo_id):
|
|
"""
|
|
Mark a todo as complete.
|
|
|
|
Args:
|
|
project_id (str): Project ID
|
|
todo_id (str): Todo ID
|
|
|
|
Returns:
|
|
dict: Completion details
|
|
"""
|
|
endpoint = f'buckets/{project_id}/todos/{todo_id}/completion.json'
|
|
response = self.post(endpoint)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to complete todo: {response.status_code} - {response.text}")
|
|
|
|
def uncomplete_todo(self, project_id, todo_id):
|
|
"""
|
|
Mark a todo as incomplete.
|
|
|
|
Args:
|
|
project_id (str): Project ID
|
|
todo_id (str): Todo ID
|
|
|
|
Returns:
|
|
bool: True if successful
|
|
"""
|
|
endpoint = f'buckets/{project_id}/todos/{todo_id}/completion.json'
|
|
response = self.delete(endpoint)
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to uncomplete todo: {response.status_code} - {response.text}")
|
|
|
|
# People methods
|
|
def get_people(self):
|
|
"""Get all people in the account."""
|
|
response = self.get('people.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get people: {response.status_code} - {response.text}")
|
|
|
|
# Campfire (chat) methods
|
|
def get_campfires(self, project_id):
|
|
"""Get the campfire for a project."""
|
|
response = self.get(f'buckets/{project_id}/chats.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get campfire: {response.status_code} - {response.text}")
|
|
|
|
def get_campfire_lines(self, project_id, campfire_id):
|
|
"""Get chat lines from a campfire."""
|
|
response = self.get(f'buckets/{project_id}/chats/{campfire_id}/lines.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get campfire lines: {response.status_code} - {response.text}")
|
|
|
|
# Message board methods
|
|
def get_message_board(self, project_id):
|
|
"""Get the message board for a project."""
|
|
response = self.get(f'projects/{project_id}/message_board.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get message board: {response.status_code} - {response.text}")
|
|
|
|
def get_messages(self, project_id):
|
|
"""Get all messages for a project."""
|
|
# First get the message board ID
|
|
message_board = self.get_message_board(project_id)
|
|
message_board_id = message_board['id']
|
|
|
|
# Then get all messages
|
|
response = self.get('messages.json', {'message_board_id': message_board_id})
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get messages: {response.status_code} - {response.text}")
|
|
|
|
# Schedule methods
|
|
def get_schedule(self, project_id):
|
|
"""Get the schedule for a project."""
|
|
response = self.get(f'projects/{project_id}/schedule.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get schedule: {response.status_code} - {response.text}")
|
|
|
|
def get_schedule_entries(self, project_id):
|
|
"""
|
|
Get schedule entries for a project.
|
|
|
|
Args:
|
|
project_id (int): Project ID
|
|
|
|
Returns:
|
|
list: Schedule entries
|
|
"""
|
|
try:
|
|
endpoint = f"buckets/{project_id}/schedules.json"
|
|
schedule = self.get(endpoint)
|
|
|
|
if isinstance(schedule, list) and len(schedule) > 0:
|
|
schedule_id = schedule[0]['id']
|
|
entries_endpoint = f"buckets/{project_id}/schedules/{schedule_id}/entries.json"
|
|
return self.get(entries_endpoint)
|
|
else:
|
|
return []
|
|
except Exception as e:
|
|
raise Exception(f"Failed to get schedule: {str(e)}")
|
|
|
|
# Comments methods
|
|
def get_comments(self, project_id, recording_id):
|
|
"""
|
|
Get all comments for a recording (todos, message, etc.).
|
|
|
|
Args:
|
|
recording_id (int): ID of the recording (todos, message, etc.)
|
|
project_id (int): Project/bucket ID. If not provided, it will be extracted from the recording ID.
|
|
|
|
Returns:
|
|
list: Comments for the recording
|
|
"""
|
|
endpoint = f"buckets/{project_id}/recordings/{recording_id}/comments.json"
|
|
response = self.get(endpoint)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get comments: {response.status_code} - {response.text}")
|
|
|
|
def create_comment(self, recording_id, bucket_id, content):
|
|
"""
|
|
Create a comment on a recording.
|
|
|
|
Args:
|
|
recording_id (int): ID of the recording to comment on
|
|
bucket_id (int): Project/bucket ID
|
|
content (str): Content of the comment in HTML format
|
|
|
|
Returns:
|
|
dict: The created comment
|
|
"""
|
|
endpoint = f"buckets/{bucket_id}/recordings/{recording_id}/comments.json"
|
|
data = {"content": content}
|
|
response = self.post(endpoint, data)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to create comment: {response.status_code} - {response.text}")
|
|
|
|
def get_comment(self, comment_id, bucket_id):
|
|
"""
|
|
Get a specific comment.
|
|
|
|
Args:
|
|
comment_id (int): Comment ID
|
|
bucket_id (int): Project/bucket ID
|
|
|
|
Returns:
|
|
dict: Comment details
|
|
"""
|
|
endpoint = f"buckets/{bucket_id}/comments/{comment_id}.json"
|
|
response = self.get(endpoint)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get comment: {response.status_code} - {response.text}")
|
|
|
|
def update_comment(self, comment_id, bucket_id, content):
|
|
"""
|
|
Update a comment.
|
|
|
|
Args:
|
|
comment_id (int): Comment ID
|
|
bucket_id (int): Project/bucket ID
|
|
content (str): New content for the comment in HTML format
|
|
|
|
Returns:
|
|
dict: Updated comment
|
|
"""
|
|
endpoint = f"buckets/{bucket_id}/comments/{comment_id}.json"
|
|
data = {"content": content}
|
|
response = self.put(endpoint, data)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to update comment: {response.status_code} - {response.text}")
|
|
|
|
def delete_comment(self, comment_id, bucket_id):
|
|
"""
|
|
Delete a comment.
|
|
|
|
Args:
|
|
comment_id (int): Comment ID
|
|
bucket_id (int): Project/bucket ID
|
|
|
|
Returns:
|
|
bool: True if successful
|
|
"""
|
|
endpoint = f"buckets/{bucket_id}/comments/{comment_id}.json"
|
|
response = self.delete(endpoint)
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to delete comment: {response.status_code} - {response.text}")
|
|
|
|
def get_daily_check_ins(self, project_id, page=1):
|
|
project = self.get_project(project_id)
|
|
questionnaire = next(_ for _ in project["dock"] if _["name"] == "questionnaire")
|
|
endpoint = f"buckets/{project_id}/questionnaires/{questionnaire['id']}/questions.json"
|
|
response = self.get(endpoint, params={"page": page})
|
|
if response.status_code != 200:
|
|
raise Exception("Failed to read questions")
|
|
return response.json()
|
|
|
|
def get_question_answers(self, project_id, question_id, page=1):
|
|
endpoint = f"buckets/{project_id}/questions/{question_id}/answers.json"
|
|
response = self.get(endpoint, params={"page": page})
|
|
if response.status_code != 200:
|
|
raise Exception("Failed to read question answers")
|
|
return response.json()
|
|
|
|
# Card Table methods
|
|
def get_card_tables(self, project_id):
|
|
"""Get all card tables for a project."""
|
|
project = self.get_project(project_id)
|
|
try:
|
|
return [item for item in project["dock"] if item.get("name") in ("kanban_board", "card_table")]
|
|
except (IndexError, TypeError):
|
|
return []
|
|
|
|
def get_card_table(self, project_id):
|
|
"""Get the first card table for a project (Basecamp 3 can have multiple card tables per project)."""
|
|
card_tables = self.get_card_tables(project_id)
|
|
if not card_tables:
|
|
raise Exception(f"No card tables found for project: {project_id}")
|
|
return card_tables[0] # Return the first card table
|
|
|
|
def get_card_table_details(self, project_id, card_table_id):
|
|
"""Get details for a specific card table."""
|
|
response = self.get(f'buckets/{project_id}/card_tables/{card_table_id}.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
elif response.status_code == 204:
|
|
# 204 means "No Content" - return an empty structure
|
|
return {"lists": [], "id": card_table_id, "status": "empty"}
|
|
else:
|
|
raise Exception(f"Failed to get card table: {response.status_code} - {response.text}")
|
|
|
|
# Card Table Column methods
|
|
def get_columns(self, project_id, card_table_id):
|
|
"""Get all columns in a card table."""
|
|
# Get the card table details which includes the lists (columns)
|
|
card_table_details = self.get_card_table_details(project_id, card_table_id)
|
|
return card_table_details.get('lists', [])
|
|
|
|
def get_column(self, project_id, column_id):
|
|
"""Get a specific column."""
|
|
response = self.get(f'buckets/{project_id}/card_tables/columns/{column_id}.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get column: {response.status_code} - {response.text}")
|
|
|
|
def create_column(self, project_id, card_table_id, title):
|
|
"""Create a new column in a card table."""
|
|
data = {"title": title}
|
|
response = self.post(f'buckets/{project_id}/card_tables/{card_table_id}/columns.json', data)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to create column: {response.status_code} - {response.text}")
|
|
|
|
def update_column(self, project_id, column_id, title):
|
|
"""Update a column title."""
|
|
data = {"title": title}
|
|
response = self.put(f'buckets/{project_id}/card_tables/columns/{column_id}.json', data)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to update column: {response.status_code} - {response.text}")
|
|
|
|
def move_column(self, project_id, column_id, position, card_table_id):
|
|
"""Move a column to a new position."""
|
|
data = {
|
|
"source_id": column_id,
|
|
"target_id": card_table_id,
|
|
"position": position
|
|
}
|
|
response = self.post(f'buckets/{project_id}/card_tables/{card_table_id}/moves.json', data)
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to move column: {response.status_code} - {response.text}")
|
|
|
|
def update_column_color(self, project_id, column_id, color):
|
|
"""Update a column color."""
|
|
data = {"color": color}
|
|
response = self.patch(f'buckets/{project_id}/card_tables/columns/{column_id}/color.json', data)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to update column color: {response.status_code} - {response.text}")
|
|
|
|
def put_column_on_hold(self, project_id, column_id):
|
|
"""Put a column on hold."""
|
|
response = self.post(f'buckets/{project_id}/card_tables/columns/{column_id}/on_hold.json')
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to put column on hold: {response.status_code} - {response.text}")
|
|
|
|
def remove_column_hold(self, project_id, column_id):
|
|
"""Remove hold from a column."""
|
|
response = self.delete(f'buckets/{project_id}/card_tables/columns/{column_id}/on_hold.json')
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to remove column hold: {response.status_code} - {response.text}")
|
|
|
|
def watch_column(self, project_id, column_id):
|
|
"""Subscribe to column notifications."""
|
|
response = self.post(f'buckets/{project_id}/card_tables/lists/{column_id}/subscription.json')
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to watch column: {response.status_code} - {response.text}")
|
|
|
|
def unwatch_column(self, project_id, column_id):
|
|
"""Unsubscribe from column notifications."""
|
|
response = self.delete(f'buckets/{project_id}/card_tables/lists/{column_id}/subscription.json')
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to unwatch column: {response.status_code} - {response.text}")
|
|
|
|
# Card Table Card methods
|
|
def get_cards(self, project_id, column_id):
|
|
"""Get all cards in a column."""
|
|
response = self.get(f'buckets/{project_id}/card_tables/lists/{column_id}/cards.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get cards: {response.status_code} - {response.text}")
|
|
|
|
def get_card(self, project_id, card_id):
|
|
"""Get a specific card."""
|
|
response = self.get(f'buckets/{project_id}/card_tables/cards/{card_id}.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get card: {response.status_code} - {response.text}")
|
|
|
|
def create_card(self, project_id, column_id, title, content=None, due_on=None, notify=False):
|
|
"""Create a new card in a column."""
|
|
data = {"title": title}
|
|
if content:
|
|
data["content"] = content
|
|
if due_on:
|
|
data["due_on"] = due_on
|
|
if notify:
|
|
data["notify"] = notify
|
|
response = self.post(f'buckets/{project_id}/card_tables/lists/{column_id}/cards.json', data)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to create card: {response.status_code} - {response.text}")
|
|
|
|
def update_card(self, project_id, card_id, title=None, content=None, due_on=None, assignee_ids=None):
|
|
"""Update a card."""
|
|
data = {}
|
|
if title:
|
|
data["title"] = title
|
|
if content:
|
|
data["content"] = content
|
|
if due_on:
|
|
data["due_on"] = due_on
|
|
if assignee_ids:
|
|
data["assignee_ids"] = assignee_ids
|
|
response = self.put(f'buckets/{project_id}/card_tables/cards/{card_id}.json', data)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to update card: {response.status_code} - {response.text}")
|
|
|
|
def move_card(self, project_id, card_id, column_id):
|
|
"""Move a card to a new column."""
|
|
data = {"column_id": column_id}
|
|
response = self.post(f'buckets/{project_id}/card_tables/cards/{card_id}/moves.json', data)
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to move card: {response.status_code} - {response.text}")
|
|
|
|
def complete_card(self, project_id, card_id):
|
|
"""Mark a card as complete."""
|
|
response = self.post(f'buckets/{project_id}/todos/{card_id}/completion.json')
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to complete card: {response.status_code} - {response.text}")
|
|
|
|
def uncomplete_card(self, project_id, card_id):
|
|
"""Mark a card as incomplete."""
|
|
response = self.delete(f'buckets/{project_id}/todos/{card_id}/completion.json')
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to uncomplete card: {response.status_code} - {response.text}")
|
|
|
|
# Card Steps methods
|
|
def get_card_steps(self, project_id, card_id):
|
|
"""Get all steps (sub-tasks) for a card."""
|
|
card = self.get_card(project_id, card_id)
|
|
return card.get('steps', [])
|
|
|
|
def create_card_step(self, project_id, card_id, title, due_on=None, assignee_ids=None):
|
|
"""Create a new step (sub-task) for a card."""
|
|
data = {"title": title}
|
|
if due_on:
|
|
data["due_on"] = due_on
|
|
if assignee_ids:
|
|
data["assignee_ids"] = assignee_ids
|
|
response = self.post(f'buckets/{project_id}/card_tables/cards/{card_id}/steps.json', data)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to create card step: {response.status_code} - {response.text}")
|
|
|
|
def get_card_step(self, project_id, step_id):
|
|
"""Get a specific card step."""
|
|
response = self.get(f'buckets/{project_id}/card_tables/steps/{step_id}.json')
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get card step: {response.status_code} - {response.text}")
|
|
|
|
def update_card_step(self, project_id, step_id, title=None, due_on=None, assignee_ids=None):
|
|
"""Update a card step."""
|
|
data = {}
|
|
if title:
|
|
data["title"] = title
|
|
if due_on:
|
|
data["due_on"] = due_on
|
|
if assignee_ids:
|
|
data["assignee_ids"] = assignee_ids
|
|
response = self.put(f'buckets/{project_id}/card_tables/steps/{step_id}.json', data)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to update card step: {response.status_code} - {response.text}")
|
|
|
|
def delete_card_step(self, project_id, step_id):
|
|
"""Delete a card step."""
|
|
response = self.delete(f'buckets/{project_id}/card_tables/steps/{step_id}.json')
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to delete card step: {response.status_code} - {response.text}")
|
|
|
|
def complete_card_step(self, project_id, step_id):
|
|
"""Mark a card step as complete."""
|
|
response = self.post(f'buckets/{project_id}/todos/{step_id}/completion.json')
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to complete card step: {response.status_code} - {response.text}")
|
|
|
|
def uncomplete_card_step(self, project_id, step_id):
|
|
"""Mark a card step as incomplete."""
|
|
response = self.delete(f'buckets/{project_id}/todos/{step_id}/completion.json')
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to uncomplete card step: {response.status_code} - {response.text}")
|
|
|
|
# New methods for additional Basecamp API functionality
|
|
def create_attachment(self, file_path, name, content_type="application/octet-stream"):
|
|
"""Upload an attachment and return the attachable sgid."""
|
|
with open(file_path, "rb") as f:
|
|
data = f.read()
|
|
|
|
headers = self.headers.copy()
|
|
headers["Content-Type"] = content_type
|
|
headers["Content-Length"] = str(len(data))
|
|
|
|
endpoint = f"attachments.json?name={name}"
|
|
response = requests.post(f"{self.base_url}/{endpoint}", auth=self.auth, headers=headers, data=data)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to create attachment: {response.status_code} - {response.text}")
|
|
|
|
def get_events(self, project_id, recording_id):
|
|
"""Get events for a recording."""
|
|
endpoint = f"buckets/{project_id}/recordings/{recording_id}/events.json"
|
|
response = self.get(endpoint)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get events: {response.status_code} - {response.text}")
|
|
|
|
def get_webhooks(self, project_id):
|
|
"""List webhooks for a project."""
|
|
endpoint = f"buckets/{project_id}/webhooks.json"
|
|
response = self.get(endpoint)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get webhooks: {response.status_code} - {response.text}")
|
|
|
|
def create_webhook(self, project_id, payload_url, types=None):
|
|
"""Create a webhook for a project."""
|
|
data = {"payload_url": payload_url}
|
|
if types:
|
|
data["types"] = types
|
|
endpoint = f"buckets/{project_id}/webhooks.json"
|
|
response = self.post(endpoint, data)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to create webhook: {response.status_code} - {response.text}")
|
|
|
|
def delete_webhook(self, project_id, webhook_id):
|
|
"""Delete a webhook."""
|
|
endpoint = f"buckets/{project_id}/webhooks/{webhook_id}.json"
|
|
response = self.delete(endpoint)
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to delete webhook: {response.status_code} - {response.text}")
|
|
|
|
def get_documents(self, project_id, vault_id):
|
|
"""List documents in a vault."""
|
|
endpoint = f"buckets/{project_id}/vaults/{vault_id}/documents.json"
|
|
response = self.get(endpoint)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get documents: {response.status_code} - {response.text}")
|
|
|
|
def get_document(self, project_id, document_id):
|
|
"""Get a single document."""
|
|
endpoint = f"buckets/{project_id}/documents/{document_id}.json"
|
|
response = self.get(endpoint)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get document: {response.status_code} - {response.text}")
|
|
|
|
def create_document(self, project_id, vault_id, title, content, status="active"):
|
|
"""Create a document in a vault."""
|
|
data = {"title": title, "content": content, "status": status}
|
|
endpoint = f"buckets/{project_id}/vaults/{vault_id}/documents.json"
|
|
response = self.post(endpoint, data)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to create document: {response.status_code} - {response.text}")
|
|
|
|
def update_document(self, project_id, document_id, title=None, content=None):
|
|
"""Update a document's title or content."""
|
|
data = {}
|
|
if title:
|
|
data["title"] = title
|
|
if content:
|
|
data["content"] = content
|
|
endpoint = f"buckets/{project_id}/documents/{document_id}.json"
|
|
response = self.put(endpoint, data)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to update document: {response.status_code} - {response.text}")
|
|
|
|
def trash_document(self, project_id, document_id):
|
|
"""Trash a document."""
|
|
endpoint = f"buckets/{project_id}/recordings/{document_id}/status/trashed.json"
|
|
response = self.put(endpoint)
|
|
if response.status_code == 204:
|
|
return True
|
|
else:
|
|
raise Exception(f"Failed to trash document: {response.status_code} - {response.text}")
|
|
|
|
# Upload methods
|
|
def get_uploads(self, project_id, vault_id=None):
|
|
"""List uploads in a project or vault."""
|
|
if vault_id:
|
|
endpoint = f"buckets/{project_id}/vaults/{vault_id}/uploads.json"
|
|
else:
|
|
endpoint = f"buckets/{project_id}/uploads.json"
|
|
response = self.get(endpoint)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get uploads: {response.status_code} - {response.text}")
|
|
|
|
def get_upload(self, project_id, upload_id):
|
|
"""Get a single upload."""
|
|
endpoint = f"buckets/{project_id}/uploads/{upload_id}.json"
|
|
response = self.get(endpoint)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"Failed to get upload: {response.status_code} - {response.text}")
|